func AppAssign(configStore *config.Store, app, env, pool string) error {
	// Don't allow deleting runtime hosts entries
	if app == "hosts" || app == "pools" {
		return fmt.Errorf("invalid app name: %s", app)
	}

	exists, err := configStore.PoolExists(env, pool)
	if err != nil {
		return err
	}

	if !exists {
		log.Warnf("WARN: Pool %s does not exist.", pool)
	}

	created, err := configStore.AssignApp(app, env, pool)

	if err != nil {
		return err
	}

	if created {
		log.Printf("Assigned %s in env %s to pool %s.\n", app, env, pool)
	} else {
		log.Printf("%s already assigned to pool %s in env %s.\n", app, pool, env)
	}
	return nil
}

func pullImage(appCfg config.App) (*docker.Image, error) {

	image, err := serviceRuntime.InspectImage(appCfg.Version())
	if image != nil && image.ID == appCfg.VersionID() || appCfg.VersionID() == "" {
		return image, nil
	}

	log.Printf("Pulling %s version %s\n", appCfg.Name(), appCfg.Version())
	image, err = serviceRuntime.PullImage(appCfg.Version(),
		appCfg.VersionID())
	if image == nil || err != nil {
		log.Errorf("ERROR: Could not pull image %s: %s",
			appCfg.Version(), err)
		return nil, err
	}

	if image.ID != appCfg.VersionID() && len(appCfg.VersionID()) > 12 {
		log.Errorf("ERROR: Pulled image for %s does not match expected ID. Expected: %s: Got: %s",
			appCfg.Version(),
			image.ID[0:12], appCfg.VersionID()[0:12])
		return nil, errors.New(fmt.Sprintf("failed to pull image ID %s", appCfg.VersionID()[0:12]))
	}

	log.Printf("Pulled %s\n", appCfg.Version())
	return image, nil
}

func (s *ServiceRuntime) stopContainer(container *docker.Container) error {
	if _, ok := blacklistedContainerId[container.ID]; ok {
		log.Printf("Container %s blacklisted. Won't try to stop.\n", container.ID)
		return nil
	}

	log.Printf("Stopping %s container %s\n", strings.TrimPrefix(container.Name, "/"), container.ID[0:12])

	c := make(chan error, 1)
	go func() { c <- s.dockerClient.StopContainer(container.ID, 10) }()
	select {
	case err := <-c:
		if err != nil {
			log.Printf("ERROR: Unable to stop container: %s\n", container.ID)
			return err
		}
	case <-time.After(20 * time.Second):
		blacklistedContainerId[container.ID] = true
		log.Printf("ERROR: Timed out trying to stop container. Zombie?. Blacklisting: %s\n", container.ID)
		return nil
	}
	log.Printf("Stopped %s container %s\n", strings.TrimPrefix(container.Name, "/"), container.ID[0:12])

	return nil
	// TODO: why is this commented out?
	//       Should we verify that containers are actually removed somehow?
	/*	return s.dockerClient.RemoveContainer(docker.RemoveContainerOptions{
		ID:            container.ID,
		RemoveVolumes: true,
	})*/
}

func poolCreate(c *cli.Context) {
	ensureEnvArg(c)
	ensurePoolArg(c)
	initStore(c)
	created, err := configStore.CreatePool(utils.GalaxyPool(c), utils.GalaxyEnv(c))
	if err != nil {
		log.Fatalf("ERROR: Could not create pool: %s", err)
		return
	}

	if created {
		log.Printf("Pool %s created\n", utils.GalaxyPool(c))
	} else {
		log.Printf("Pool %s already exists\n", utils.GalaxyPool(c))
	}

	ec2host, err := runtime.EC2PublicHostname()
	if err != nil || ec2host == "" {
		log.Debug("not running from AWS, skipping pool creation")
		return
	}

	// now create the cloudformation stack
	// is this fails, the stack can be created separately with
	// stack:create_pool
	stackCreatePool(c)
}

func AppDeploy(configStore *config.Store, serviceRuntime *runtime.ServiceRuntime, app, env, version string) error {
	log.Printf("Pulling image %s...", version)

	image, err := serviceRuntime.PullImage(version, "")
	if image == nil || err != nil {
		return fmt.Errorf("unable to pull %s. Has it been released yet?", version)
	}

	svcCfg, err := configStore.GetApp(app, env)
	if err != nil {
		return fmt.Errorf("unable to deploy app: %s.", err)
	}

	if svcCfg == nil {
		return fmt.Errorf("app %s does not exist. Create it first.", app)
	}

	svcCfg.SetVersion(version)
	svcCfg.SetVersionID(utils.StripSHA(image.ID))

	updated, err := configStore.UpdateApp(svcCfg, env)
	if err != nil {
		return fmt.Errorf("could not store version: %s", err)
	}
	if !updated {
		return fmt.Errorf("%s NOT deployed.", version)
	}
	log.Printf("Deployed %s.\n", version)
	return nil
}

func ConfigList(configStore *config.Store, app, env string) error {

	cfg, err := configStore.GetApp(app, env)
	if err != nil {
		return err
	}

	if cfg == nil {
		return fmt.Errorf("unable to list config for %s.", app)
	}

	keys := sort.StringSlice{"ENV"}
	for k, _ := range cfg.Env() {
		keys = append(keys, k)
	}

	keys.Sort()

	for _, k := range keys {
		if k == "ENV" {
			log.Printf("%s=%s\n", k, env)
			continue
		}
		log.Printf("%s=%s\n", k, cfg.Env()[k])
	}

	return nil
}

func pgPsql(c *cli.Context) {
	ensureEnvArg(c)
	initStore(c)
	app := ensureAppParam(c, "pg:psql")

	appCfg, err := configStore.GetApp(app, utils.GalaxyEnv(c))
	if err != nil {
		log.Fatalf("ERROR: Unable to run command: %s.", err)
		return
	}

	database_url := appCfg.Env()["DATABASE_URL"]
	if database_url == "" {
		log.Printf("No DATABASE_URL configured.  Set one with config:set first.")
		return
	}

	if !strings.HasPrefix(database_url, "postgres://") {
		log.Printf("DATABASE_URL is not a postgres database.")
		return
	}

	cmd := exec.Command("psql", database_url)

	cmd.Stdin = os.Stdin
	cmd.Stdout = os.Stdout
	cmd.Stderr = os.Stderr

	// Ignore SIGINT while the process is running
	ch := make(chan os.Signal, 1)
	signal.Notify(ch, os.Interrupt)

	defer func() {
		signal.Stop(ch)
		close(ch)
	}()

	go func() {
		for {
			_, ok := <-ch
			if !ok {
				break
			}
		}
	}()

	err = cmd.Start()
	if err != nil {
		log.Fatal(err)
	}

	err = cmd.Wait()
	if err != nil {
		fmt.Printf("Command finished with error: %v\n", err)
	}
}

func monitorService(changedConfigs chan *config.ConfigChange) {

	for {

		var changedConfig *config.ConfigChange
		select {

		case changedConfig = <-changedConfigs:

			if changedConfig.Error != nil {
				log.Errorf("ERROR: Error watching changes: %s", changedConfig.Error)
				continue
			}

			if changedConfig.AppConfig == nil {
				continue
			}

			assigned, err := appAssigned(changedConfig.AppConfig.Name())
			if err != nil {
				log.Errorf("ERROR: Error retrieving service config for %s: %s", changedConfig.AppConfig.Name(), err)
				if !loop {
					return
				}
				continue
			}

			if !assigned {
				continue
			}

			ch, ok := workerChans[changedConfig.AppConfig.Name()]
			if !ok {
				name := changedConfig.AppConfig.Name()
				ch := make(chan string)
				workerChans[name] = ch
				wg.Add(1)
				go restartContainers(name, ch)
				ch <- "deploy"

				log.Printf("Started new worker for %s\n", name)
				continue
			}

			if changedConfig.Restart {
				log.Printf("Restarting %s", changedConfig.AppConfig.Name())
				ch <- "restart"
			} else {
				ch <- "deploy"
			}
		}
	}

}

func (r *RedisBackend) subscribeChannel(key string, msgs chan string) {
	var wg sync.WaitGroup

	redisPool := redis.Pool{
		MaxIdle:     1,
		IdleTimeout: 0,
		Dial: func() (redis.Conn, error) {
			return redis.DialTimeout("tcp", r.RedisHost, time.Second, 0, 0)
		},
		// test every connection for now
		TestOnBorrow: r.testOnBorrow,
	}

	for {
		conn := redisPool.Get()
		// no defer, doesn't return
		if err := conn.Err(); err != nil {
			conn.Close()
			log.Printf("ERROR: %v\n", err)
			time.Sleep(5 * time.Second)
			continue
		}

		wg.Add(1)
		psc := redis.PubSubConn{Conn: conn}
		go func() {
			defer wg.Done()
			for {
				switch n := psc.Receive().(type) {
				case redis.Message:
					msg := string(n.Data)
					msgs <- msg
				case error:
					psc.Close()
					log.Printf("ERROR: %v\n", n)
					return
				}
			}
		}()

		wg.Add(1)
		go func() {
			defer wg.Done()
			psc.Subscribe(key)
			log.Printf("Monitoring for config changes on channel: %s\n", key)
		}()
		wg.Wait()
		conn.Close()
	}
}

func Register(serviceRuntime *runtime.ServiceRuntime, configStore *config.Store, env, pool, hostIP, shuttleAddr string) {
	if shuttleAddr != "" {
		client = shuttle.NewClient(shuttleAddr)
	}

	RegisterAll(serviceRuntime, configStore, env, pool, hostIP, shuttleAddr, false)

	containerEvents := make(chan runtime.ContainerEvent)
	err := serviceRuntime.RegisterEvents(env, pool, hostIP, containerEvents)
	if err != nil {
		log.Printf("ERROR: Unable to register docker event listener: %s", err)
	}

	for {

		select {
		case ce := <-containerEvents:
			switch ce.Status {
			case "start":
				reg, err := configStore.RegisterService(env, pool, hostIP, ce.Container)
				if err != nil {
					log.Errorf("ERROR: Unable to register container: %s", err)
					continue
				}

				log.Printf("Registered %s running as %s for %s%s", strings.TrimPrefix(reg.ContainerName, "/"),
					reg.ContainerID[0:12], reg.Name, locationAt(reg))
				registerShuttle(configStore, env, shuttleAddr)
			case "die", "stop":
				reg, err := configStore.UnRegisterService(env, pool, hostIP, ce.Container)
				if err != nil {
					log.Errorf("ERROR: Unable to unregister container: %s", err)
					continue
				}

				if reg != nil {
					log.Printf("Unregistered %s running as %s for %s%s", strings.TrimPrefix(reg.ContainerName, "/"),
						reg.ContainerID[0:12], reg.Name, locationAt(reg))
				}
				RegisterAll(serviceRuntime, configStore, env, pool, hostIP, shuttleAddr, true)
				pruneShuttleBackends(configStore, env, shuttleAddr)
			}

		case <-time.After(10 * time.Second):
			RegisterAll(serviceRuntime, configStore, env, pool, hostIP, shuttleAddr, true)
			pruneShuttleBackends(configStore, env, shuttleAddr)
		}
	}
}

func poolCreate(c *cli.Context) {
	ensureEnvArg(c)
	ensurePoolArg(c)
	initStore(c)
	created, err := configStore.CreatePool(utils.GalaxyPool(c), utils.GalaxyEnv(c))
	if err != nil {
		log.Fatalf("ERROR: Could not create pool: %s", err)
		return
	}

	if created {
		log.Printf("Pool %s created\n", utils.GalaxyPool(c))
	} else {
		log.Printf("Pool %s already exists\n", utils.GalaxyPool(c))
	}
}

func (s *ServiceRuntime) RegisterAll(env, pool, hostIP string) ([]*config.ServiceRegistration, error) {
	// make sure any old containers that shouldn't be running are gone
	// FIXME: I don't like how a "Register" function has the possible side
	//        effect of stopping containers
	s.StopUnassigned(env, pool)

	containers, err := s.ManagedContainers()
	if err != nil {
		return nil, err
	}

	registrations := []*config.ServiceRegistration{}

	for _, container := range containers {
		name := s.EnvFor(container)["GALAXY_APP"]

		registration, err := s.configStore.RegisterService(env, pool, hostIP, container)
		if err != nil {
			log.Printf("ERROR: Could not register %s: %s\n", name, err.Error())
			continue
		}
		registrations = append(registrations, registration)
	}

	return registrations, nil

}

func poolDelete(c *cli.Context) {
	ensureEnvArg(c)
	ensurePoolArg(c)
	initStore(c)
	empty, err := configStore.DeletePool(utils.GalaxyPool(c), utils.GalaxyEnv(c))
	if err != nil {
		log.Fatalf("ERROR: Could not delete pool: %s", err)
		return
	}

	if empty {
		log.Printf("Pool %s deleted\n", utils.GalaxyPool(c))

	} else {
		log.Printf("Pool %s has apps assigned. Unassign them first.\n", utils.GalaxyPool(c))
	}
}

func AppUnassign(configStore *config.Store, app, env, pool string) error {
	// Don't allow deleting runtime hosts entries
	if app == "hosts" || app == "pools" {
		return fmt.Errorf("invalid app name: %s", app)
	}

	deleted, err := configStore.UnassignApp(app, env, pool)
	if err != nil {
		return err
	}

	if deleted {
		log.Printf("Unassigned %s in env %s from pool %s\n", app, env, pool)
	} else {
		log.Printf("%s could not be unassigned.\n", pool)
	}
	return nil
}

// return --base, or try to find a base cloudformation stack
func getBase(c *cli.Context) string {
	errNoBase := fmt.Errorf("could not identify a unique base stack")

	base := c.String("base")
	if base != "" {
		return base
	}

	descResp, err := stack.DescribeStacks("")
	if err != nil {
		log.Fatal(err)
	}

	for _, stack := range descResp.Stacks {
		// first check for galaxy:base tag
		baseTag := false
		for _, t := range stack.Tags {
			if t.Key == "galaxy" && t.Value == "base" {
				baseTag = true
			}
		}
		if baseTag {
			if base != "" {
				err = errNoBase
			}
			base = stack.Name
			continue
		}
		parts := strings.Split(stack.Name, "-")

		// check for "-base" in the name
		if parts[len(parts)-1] == "base" {
			if base != "" {
				err = errNoBase
			}
			base = stack.Name
			continue
		}

		// the best we can do for now is look for a stack with a single word
		if len(parts) == 1 {
			if base != "" {
				err = errNoBase
			}
			base = stack.Name

			log.Printf("Warning: guessing base stack: %s", base)
		}
	}

	if err != nil {
		log.Fatalf("%s: %s", err, "use --base")
	}

	return base
}

func pullImage(appCfg config.App) (*docker.Image, error) {
	image, err := serviceRuntime.PullImage(appCfg.Version(), appCfg.VersionID())
	if image == nil || err != nil {
		log.Errorf("ERROR: Could not pull image %s: %s", appCfg.Version(), err)
		return nil, err
	}

	log.Printf("Pulled %s version %s\n", appCfg.Name(), appCfg.Version())
	return image, nil
}

func AppCreate(configStore *config.Store, app, env string) error {
	// Don't allow creating runtime hosts entries
	if app == "hosts" {
		return fmt.Errorf("could not create app: %s", app)
	}

	created, err := configStore.CreateApp(app, env)

	if err != nil {
		return fmt.Errorf("could not create app: %s", err)
	}

	if created {
		log.Printf("Created %s in env %s.\n", app, env)
	} else {
		log.Printf("%s already exists in in env %s.", app, env)
	}
	return nil
}

func AppDelete(configStore *config.Store, app, env string) error {

	// Don't allow deleting runtime hosts entries
	if app == "hosts" || app == "pools" {
		return fmt.Errorf("could not delete app: %s", app)
	}

	deleted, err := configStore.DeleteApp(app, env)
	if err != nil {
		return fmt.Errorf("could not delete app: %s", err)
	}

	if deleted {
		log.Printf("Deleted %s from env %s.\n", app, env)
	} else {
		log.Printf("%s does not exists in env %s.\n", app, env)
	}
	return nil
}

// inspectImage checks that the running image matches the config.
// We only use this to print warnings, since we likely need to deploy a new
// config version to fix the inconsistency.
func inspectImage(appCfg config.App) {
	image, err := serviceRuntime.InspectImage(appCfg.Version())
	if err != nil {
		log.Println("error inspecting image", appCfg.Version())
		return
	}

	if utils.StripSHA(image.ID) != appCfg.VersionID() {
		log.Printf("warning: %s image ID does not match config", appCfg.Name())
	}
}

func ConfigUnset(configStore *config.Store, app, env string, envVars []string) error {

	if len(envVars) == 0 {
		return fmt.Errorf("no config values specified.")
	}

	svcCfg, err := configStore.GetApp(app, env)
	if err != nil {
		return fmt.Errorf("unable to unset config: %s.", err)
	}

	updated := false
	for _, arg := range envVars {
		k := strings.ToUpper(strings.TrimSpace(arg))
		if k == "ENV" || svcCfg.EnvGet(k) == "" {
			log.Warnf("%s cannot be unset.", k)
			continue
		}

		log.Printf("%s\n", k)
		svcCfg.EnvSet(strings.ToUpper(arg), "")
		updated = true
	}

	if !updated {
		return fmt.Errorf("Configuration NOT changed for %s", app)
	}

	updated, err = configStore.UpdateApp(svcCfg, env)
	if err != nil {
		return fmt.Errorf("ERROR: Unable to unset config: %s.", err)

	}

	if !updated {
		return fmt.Errorf("Configuration NOT changed for %s", app)

	}
	log.Printf("Configuration changed for %s. v%d.\n", app, svcCfg.ID())
	return nil
}