func main() {
	kingpin.Version(version.VERSION)
	_, opts, _ := flags.ParseWithConsulOptions()
	client := kp.NewConsulClient(opts)
	store := kp.NewConsulStore(client)
	podStore := podstore.NewConsul(client.KV())

	if *nodeName == "" {
		hostname, err := os.Hostname()
		if err != nil {
			log.Fatalf("Could not get the hostname to do scheduling: %s", err)
		}
		*nodeName = hostname
	}

	if *manifestPath == "" {
		kingpin.Usage()
		log.Fatalln("No manifest given")
	}

	podManifest, err := manifest.FromPath(*manifestPath)
	if err != nil {
		log.Fatalf("Could not read manifest at %s: %s\n", *manifestPath, err)
	}

	out := schedule.Output{
		PodID: podManifest.ID(),
	}
	if *uuidPod {
		out.PodUniqueKey, err = podStore.Schedule(podManifest, types.NodeName(*nodeName))
		if err != nil {
			log.Fatalf("Could not schedule pod: %s", err)
		}
	} else {

		// Legacy pod
		podPrefix := kp.INTENT_TREE
		if *hookGlobal {
			podPrefix = kp.HOOK_TREE
		}
		_, err := store.SetPod(podPrefix, types.NodeName(*nodeName), podManifest)
		if err != nil {
			log.Fatalf("Could not write manifest %s to intent store: %s\n", podManifest.ID(), err)
		}
	}

	outBytes, err := json.Marshal(out)
	if err != nil {
		log.Fatalf("Successfully scheduled manifest but couldn't marshal JSON output")
	}

	fmt.Println(string(outBytes))
}

func TestStartWatchBasic(t *testing.T) {
	quitCh := make(chan struct{})
	defer close(quitCh)
	hs, checker := NewFakeHealthStore()
	go hs.StartWatch(quitCh)

	node := types.NodeName("abc01.sjc1")
	podID1 := types.PodID("podID1")
	podID2 := types.PodID("podID2")

	result := hs.Fetch(podID1, node)
	if result != nil {
		t.Errorf("expected cache to start empty, found %v", result)
	}

	checker.Send([]*health.Result{
		{ID: podID1, Node: node},
		{ID: podID2, Node: node},
	})

	result = hs.Fetch(podID1, node)
	if result == nil {
		t.Errorf("expected health store to have %s", podID1)
	}

	result = hs.Fetch(podID2, node)
	if result == nil {
		t.Errorf("expected health store to have %s", podID2)
	}
}

func TestWriteRealityIndex(t *testing.T) {
	node := types.NodeName("some_node")
	key := types.NewPodUUID()

	realityIndexPath := fmt.Sprintf("reality/%s/%s", node, key.ID)

	store, fakeKV := storeWithFakeKV(t, make(map[string]Pod), make(map[string]PodIndex))

	// confirm that the reality index doesn't exist
	pair, _, err := fakeKV.Get(realityIndexPath, nil)
	if err != nil {
		t.Fatalf("Initial conditions were not met: error fetching %s: %s", realityIndexPath, err)
	}

	if pair != nil {
		t.Fatalf("Initial conditions were not met: expected key %s to not exist", realityIndexPath)
	}

	err = store.WriteRealityIndex(key, node)
	if err != nil {
		t.Fatal(err)
	}

	pair, _, err = fakeKV.Get(realityIndexPath, nil)
	if err != nil {
		t.Fatalf("Unable to fetch the deleted key (%s): %s", realityIndexPath, err)
	}

	if pair == nil {
		t.Fatalf("%s should have been written but it wasn't", realityIndexPath)
	}
}

func TestFakeServiceHealth(t *testing.T) {
	fake := FakePodStore{}
	targetService := "paladin"
	targetHost := types.NodeName("aaa2.dfw.square")
	targetStatus := "healthy"

	fake.healthResults = map[string]kp.WatchResult{
		kp.HealthPath("shrimpy", "aaa1.dfw.square"): kp.WatchResult{
			Service: "shrimpy",
			Status:  "critical",
		},
		kp.HealthPath(targetService, targetHost): kp.WatchResult{
			Service: targetService,
			Status:  targetStatus,
		},
	}

	serviceRes, err := fake.GetServiceHealth(targetService)
	if err != nil {
		t.Fatal(err)
	}
	if len(serviceRes) != 1 {
		t.Fatalf("Expected %v to have a single health entry, found %v", targetService, len(serviceRes))
	}
	watchResult, ok := serviceRes[kp.HealthPath(targetService, targetHost)]
	if !ok {
		t.Fatalf("Expected to find a result for %v", targetHost)
	}
	if watchResult.Status != targetStatus {
		t.Fatalf("Status didn't match expected: %v", watchResult.Status)
	}
}

// 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

}

func schedule(client client.Client, logger logging.Logger) {
	m, err := manifest.FromPath(*manifestFile)
	if err != nil {
		logger.Fatalf("Could not read manifest: %s", err)
	}

	podUniqueKey, err := client.Schedule(m, types.NodeName(*node))
	if err != nil {
		logger.Fatalf("Could not schedule: %s", err)
	}

	output := struct {
		PodID        types.PodID        `json:"pod_id"`
		PodUniqueKey types.PodUniqueKey `json:"pod_unique_key"`
	}{
		PodID:        m.ID(),
		PodUniqueKey: podUniqueKey,
	}

	outBytes, err := json.Marshal(output)
	if err != nil {
		logger.Infof("Scheduled pod with key: %s", podUniqueKey)
		return
	}

	fmt.Println(string(outBytes))
}

// Returns nodes to be removed and nodes to be added
func makeNodeChanges(oldNodeLabels []labels.Labeled, newNodeLabels []labels.Labeled) ([]types.NodeName, []types.NodeName) {
	var oldNodeNames []types.NodeName
	var newNodeNames []types.NodeName

	for _, node := range oldNodeLabels {
		oldNodeNames = append(oldNodeNames, types.NodeName(node.ID))
	}

	for _, node := range newNodeLabels {
		newNodeNames = append(newNodeNames, types.NodeName(node.ID))
	}

	toRemove := types.NewNodeSet(oldNodeNames...).Difference(types.NewNodeSet(newNodeNames...)).ListNodes()
	toAdd := types.NewNodeSet(newNodeNames...).Difference(types.NewNodeSet(oldNodeNames...)).ListNodes()

	return toRemove, toAdd
}

func TestSchedule(t *testing.T) {
	store, kv := storeWithFakeKV(t, nil, nil)

	node := types.NodeName("some_node")
	key, err := store.Schedule(testManifest(), node)
	if err != nil {
		t.Fatalf("Unexpected error scheduling pod: %s", err)
	}

	// Now reach into the consul impl to make sure it was stored with an index
	// and a main pod
	podPath := fmt.Sprintf("pods/%s", key.ID)
	indexPath := fmt.Sprintf("intent/%s/%s", node, key.ID)

	if kv.Entries[podPath] == nil {
		t.Fatalf("Key '%s' wasn't set as expected", podPath)
	}

	var pod Pod
	err = json.Unmarshal(kv.Entries[podPath].Value, &pod)
	if err != nil {
		t.Fatal(err)
	}

	if pod.Node != node {
		t.Errorf("Pod wasn't stored with correct node, wanted '%s' was '%s'", node, pod.Node)
	}

	expectedManifestSha, err := testManifest().SHA()
	if err != nil {
		t.Fatal(err)
	}

	podManifestSha, err := pod.Manifest.SHA()
	if err != nil {
		t.Fatal(err)
	}

	if expectedManifestSha != podManifestSha {
		t.Error("Pod was scheduled with the wrong manifest")
	}

	if kv.Entries[indexPath] == nil {
		t.Fatalf("Index '%s' wasn't set as expected", indexPath)
	}

	var index PodIndex
	err = json.Unmarshal(kv.Entries[indexPath].Value, &index)
	if err != nil {
		t.Fatal(err)
	}

	if index.PodKey.ID != key.ID {
		t.Errorf("Index didn't have expected key, wanted '%s' was '%s'", key.ID, index.PodKey)
	}
}

func scheduleForThisHost(manifest manifest.Manifest, alsoReality bool) error {
	store := kp.NewConsulStore(kp.NewConsulClient(kp.Options{
		Token: *consulToken,
	}))
	hostname, err := os.Hostname()
	if err != nil {
		return err
	}
	_, err = store.SetPod(kp.INTENT_TREE, types.NodeName(hostname), manifest)
	if err != nil {
		return err
	}

	if alsoReality {
		_, err = store.SetPod(kp.REALITY_TREE, types.NodeName(hostname), manifest)
		return err
	}
	return nil
}

func (sel *applicatorScheduler) EligibleNodes(_ manifest.Manifest, selector klabels.Selector) ([]types.NodeName, error) {
	nodes, err := sel.applicator.GetMatches(selector, labels.NODE, false)
	if err != nil {
		return nil, err
	}

	result := make([]types.NodeName, len(nodes))
	for i, node := range nodes {
		result[i] = types.NodeName(node.ID)
	}
	return result, nil
}

func NodeAndPodIDFromPodLabel(labeled Labeled) (types.NodeName, types.PodID, error) {
	if labeled.LabelType != POD {
		return "", "", util.Errorf("Label was not a pod label, was %s", labeled.LabelType)
	}

	parts := strings.SplitN(labeled.ID, "/", 2)
	if len(parts) < 2 {
		return "", "", util.Errorf("malformed pod label %s", labeled.ID)
	}

	return types.NodeName(parts[0]), types.PodID(parts[1]), nil
}

// Initializes a pod from the current_manifest.yaml file in the pod's home directory. This function
// will error or return an old manifest if run during an inappropriate hook event, use of this
// function is discouraged in most cases
func (h *HookEnv) PodFromDisk() (*pods.Pod, error) {
	node, err := h.Node()
	if err != nil {
		return nil, err
	}
	podHome, err := h.PodHome()
	if err != nil {
		return nil, err
	}

	return pods.PodFromPodHome(types.NodeName(node), podHome)
}

func updateWithUniformHealth(t *testing.T, numNodes int, status health.HealthState) (update, map[types.NodeName]health.Result) {
	current := map[types.NodeName]bool{}
	checks := map[types.NodeName]health.Result{}

	for i := 0; i < numNodes; i++ {
		node := types.NodeName(fmt.Sprintf("node%d", i))
		current[node] = true
		checks[node] = health.Result{Status: status}
	}

	upd, _, _ := updateWithHealth(t, numNodes, 0, current, nil, nil)
	return upd, checks
}

func (h *HookEnv) Node() (types.NodeName, error) {
	node := os.Getenv(HOOKED_NODE_ENV_VAR)
	if node == "" {
		// TODO: This can't be a hard error right now. It would mean hooks built with this
		// change couldn't run under preparers without this change. We'll fix it later.
		// (At the present time, hooks shouldn't care about their node name.)
		hostname, err := os.Hostname()
		if err != nil {
			return "", util.Errorf("Error getting node name: %v", err)
		}
		node = hostname
	}
	return types.NodeName(node), nil
}

func (hs *healthStore) cache(results []*health.Result) {
	tmpCache := make(map[cacheKey]*health.Result, len(results))

	// duplicate key is undefined behavior
	for _, result := range results {
		result := result
		tmpCache[newCacheKey(result.ID, types.NodeName(result.Node))] = result
	}

	hs.lock.Lock()
	defer hs.lock.Unlock()

	hs.cachedHealth = tmpCache
}

func main() {
	// CLI takes a hostname, a token and recursively deletes all pods
	// in the reality tree. This is useful if any pods on a host have
	// been manually altered in some way and need to be restored to
	// a known state.
	kingpin.Version(version.VERSION)
	_, opts := flags.ParseWithConsulOptions()

	client := kp.NewConsulClient(opts)
	store := kp.NewConsulStore(client)

	pods, _, err := store.ListPods(kp.REALITY_TREE, types.NodeName(*nodeName))
	if err != nil {
		log.Fatalf("Could not list pods for node %v: %v", *nodeName, err)
	}
	for _, pod := range pods {
		log.Printf("Deleting %v from reality\n", pod.Manifest.ID())
		_, err := store.DeletePod(kp.REALITY_TREE, types.NodeName(*nodeName), pod.Manifest.ID())
		if err != nil {
			log.Fatalf("Could not remove %v from pod reality tree: %v", err)
		}
	}
}

func TestReadPodFromIndex(t *testing.T) {
	node := types.NodeName("some_node")
	key := types.NewPodUUID()

	podPath := fmt.Sprintf("pods/%s", key.ID)
	indexPath := fmt.Sprintf("intent/%s/%s", node, key.ID)

	pod := Pod{
		Manifest: testManifest(),
		Node:     node,
	}

	index := PodIndex{
		PodKey: key,
	}

	// Initialize the store with entries at the pod path and index path
	pods := map[string]Pod{
		podPath: pod,
	}

	// This test doesn't actually care if there's an index, but let's keep it hygienic
	indices := map[string]PodIndex{
		indexPath: index,
	}
	store, _ := storeWithFakeKV(t, pods, indices)

	outPod, err := store.ReadPod(key)
	if err != nil {
		t.Fatalf("Unexpected error reading pod: %s", err)
	}

	if pod.Node != outPod.Node {
		t.Errorf("Pod node, didn't match expected, wanted %+v was %+v", pod.Node, outPod.Node)
	}

	expectedManifestSha, err := testManifest().SHA()
	if err != nil {
		t.Fatal(err)
	}

	podManifestSha, err := outPod.Manifest.SHA()
	if err != nil {
		t.Fatal(err)
	}

	if expectedManifestSha != podManifestSha {
		t.Error("Pod returned with the wrong manifest")
	}
}

func main() {
	kingpin.Version(version.VERSION)
	_, opts := flags.ParseWithConsulOptions()

	if *nodeName == "" {
		hostname, err := os.Hostname()
		if err != nil {
			fmt.Fprintf(os.Stderr, "error getting hostname. use --node to specify a node: %v\n", err)
			os.Exit(1)
		}
		*nodeName = hostname
	}

	rm := NewP2RM(kp.NewConsulClient(opts), *podName, types.NodeName(*nodeName))

	podIsManagedByRC, rcID, err := rm.checkForManagingReplicationController()
	if err != nil {
		os.Exit(2)
	}

	if !podIsManagedByRC {
		err = rm.deletePod()
		if err != nil {
			os.Exit(2)
		}
	}

	if podIsManagedByRC && !*deallocation {
		fmt.Fprintf(
			os.Stderr,
			"error: %s is managed by replication controller: %s\n"+
				"It's possible you meant you deallocate this pod on this node. If so, please confirm your intention with --deallocate\n", *nodeName, rcID)
		os.Exit(2)
	}

	if podIsManagedByRC && *deallocation {
		err = rm.decrementDesiredCount(rcID)
		if err != nil {
			fmt.Fprintf(os.Stderr,
				"Encountered error deallocating from the RC %s. You may attempt this command again or use `p2-rctl` to cleanup manually.\n%v",
				rcID,
				err)
		}
	}

	fmt.Printf("%s: successfully removed %s\n", rm.NodeName, rm.PodName)
}

func handlePodRemoval(consulClient consulutil.ConsulClient, labeler labels.ApplicatorWithoutWatches) error {
	var rm *P2RM
	if *podUniqueKey != "" {
		rm = NewUUIDP2RM(consulClient, types.PodUniqueKey(*podUniqueKey), types.PodID(*podName), labeler)
	} else {
		if *nodeName == "" {
			hostname, err := os.Hostname()
			if err != nil {
				return fmt.Errorf("error getting hostname. use --node to specify a node: %v\n", err)
			}
			*nodeName = hostname
		}

		rm = NewLegacyP2RM(consulClient, types.PodID(*podName), types.NodeName(*nodeName), labeler)
	}

	podIsManagedByRC, rcID, err := rm.checkForManagingReplicationController()
	if err != nil {
		return err
	}

	if !podIsManagedByRC {
		err = rm.deletePod()
		if err != nil {
			return err
		}
	}

	if podIsManagedByRC && !*deallocation {
		return fmt.Errorf("error: %s is managed by replication controller: %s\n"+
			"It's possible you meant you deallocate this pod on this node. If so, please confirm your intention with --deallocate\n", *nodeName, rcID)
	}

	if podIsManagedByRC && *deallocation {
		err = rm.decrementDesiredCount(rcID)
		if err != nil {
			return fmt.Errorf("Encountered error deallocating from the RC %s. You may attempt this command again or use `p2-rctl` to cleanup manually.\n%v",
				rcID,
				err)
		}
	}

	fmt.Printf("%s: successfully removed %s\n", rm.NodeName, rm.PodID)
	return nil
}

func verifyHealthChecks(config *preparer.PreparerConfig, services []string) error {
	client, err := config.GetConsulClient()
	if err != nil {
		return err
	}
	store := kp.NewConsulStore(client)

	time.Sleep(30 * time.Second)
	// check consul for health information for each app
	name, err := os.Hostname()
	if err != nil {
		return err
	}

	node := types.NodeName(name)
	for _, sv := range services {
		res, err := store.GetHealth(sv, node)
		if err != nil {
			return err
		} else if (res == kp.WatchResult{}) {
			return fmt.Errorf("No results for %s: \n\n %s", sv, targetLogs())
		} else if res.Status != string(health.Passing) {
			return fmt.Errorf("%s did not pass health check: \n\n %s", sv, targetLogs())
		} else {
			fmt.Println(res)
		}
	}

	for _, sv := range services {
		res, err := store.GetServiceHealth(sv)
		getres, _ := store.GetHealth(sv, node)
		if err != nil {
			return err
		}
		val := res[kp.HealthPath(sv, node)]
		if getres.Id != val.Id || getres.Service != val.Service || getres.Status != val.Status {
			return fmt.Errorf("GetServiceHealth failed %+v: \n\n%s", res, targetLogs())
		}
	}

	// if it reaches here it means health checks
	// are being written to the KV store properly
	return nil
}