func flatten(imageID string) error {
	config := docker.Config{
		Image:        imageID,
		Cmd:          []string{"/bin/bash"},
		AttachStdin:  false,
		AttachStdout: false,
		AttachStderr: false,
	}
	_, c, err := dockerCluster().CreateContainer(&config)
	if err != nil {
		return err
	}
	buf := &bytes.Buffer{}
	if err := dockerCluster().ExportContainer(c.ID, buf); err != nil {
		log.Errorf("Flatten: Caugh error while exporting container %s: %s", c.ID, err.Error())
		return err
	}
	out := &bytes.Buffer{}
	opts := dcli.ImportImageOptions{Repository: imageID, Source: "-"}
	if err := dockerCluster().ImportImage(opts, buf, out); err != nil {
		log.Errorf("Flatten: Caugh error while importing image from container %s: %s", c.ID, err.Error())
		return err
	}
	if err := dockerCluster().RemoveContainer(c.ID); err != nil {
		log.Errorf("Flatten: Caugh error while removing container %s: %s", c.ID, err.Error())
	}
	removeFromRegistry(imageID)
	return nil
}

func (r hipacheRouter) AddRoute(name, address string) error {
	backendName, err := router.Retrieve(name)
	if err != nil {
		return err
	}
	domain, err := config.GetString("hipache:domain")
	if err != nil {
		log.Errorf("error on getting hipache domain in add route for %s - %s", backendName, address)
		return &routeError{"add", err}
	}
	frontend := "frontend:" + backendName + "." + domain
	if err := r.addRoute(frontend, address); err != nil {
		log.Errorf("error on add route for %s - %s", backendName, address)
		return &routeError{"add", err}
	}
	cname, err := r.getCName(backendName)
	if err != nil {
		log.Errorf("error on get cname in add route for %s - %s", backendName, address)
		return err
	}
	if cname == "" {
		return nil
	}
	return r.addRoute("frontend:"+cname, address)
}

// replicateImage replicates the given image through all nodes in the cluster.
func replicateImage(name string) error {
	var buf safe.Buffer
	if _, err := config.GetString("docker:registry"); err == nil {
		pushOpts := docker.PushImageOptions{Name: name, OutputStream: &buf}
		for i := 0; i < maxTry; i++ {
			err = dockerCluster().PushImage(pushOpts, docker.AuthConfiguration{})
			if err == nil {
				buf.Reset()
				break
			}
			log.Errorf("[docker] Failed to push image %q (%s): %s", name, err, buf.String())
			buf.Reset()
		}
		if err != nil {
			return err
		}
		pullOpts := docker.PullImageOptions{Repository: name, OutputStream: &buf}
		for i := 0; i < maxTry; i++ {
			err = dockerCluster().PullImage(pullOpts)
			if err == nil {
				break
			}
			buf.Reset()
		}
		if err != nil {
			log.Errorf("[docker] Failed to replicate image %q through nodes (%s): %s", name, err, buf.String())
			return err
		}
	}
	return nil
}

func deploy(app provision.App, version string, w io.Writer) (string, error) {
	commands, err := deployCmds(app, version)
	if err != nil {
		return "", err
	}
	imageId := getImage(app)
	actions := []*action.Action{&createContainer, &startContainer, &insertContainer}
	pipeline := action.NewPipeline(actions...)
	err = pipeline.Execute(app, imageId, commands)
	if err != nil {
		log.Errorf("error on execute deploy pipeline for app %s - %s", app.GetName(), err)
		return "", err
	}
	c := pipeline.Result().(container)
	err = c.logs(w)
	if err != nil {
		log.Errorf("error on get logs for container %s - %s", c.ID, err)
		return "", err
	}
	_, err = dockerCluster().WaitContainer(c.ID)
	if err != nil {
		log.Errorf("Process failed for container %q: %s", c.ID, err)
		return "", err
	}
	imageId, err = c.commit()
	if err != nil {
		log.Errorf("error on commit container %s - %s", c.ID, err)
		return "", err
	}
	c.remove()
	return imageId, nil
}

// replicateImage replicates the given image through all nodes in the cluster.
func replicateImage(name string) error {
	var buf bytes.Buffer
	if registry, err := config.GetString("docker:registry"); err == nil {
		if !strings.HasPrefix(name, registry) {
			name = registry + "/" + name
		}
		pushOpts := dclient.PushImageOptions{Name: name}
		for i := 0; i < maxTry; i++ {
			err = dockerCluster().PushImage(pushOpts, dclient.AuthConfiguration{}, &buf)
			if err == nil {
				buf.Reset()
				break
			}
			log.Errorf("[docker] Failed to push image %q (%s): %s", name, err, buf.String())
			buf.Reset()
		}
		if err != nil {
			return err
		}
		pullOpts := dclient.PullImageOptions{Repository: name}
		for i := 0; i < maxTry; i++ {
			err = dockerCluster().PullImage(pullOpts, &buf)
			if err == nil {
				break
			}
			buf.Reset()
		}
		if err != nil {
			log.Errorf("[docker] Failed to replicate image %q through nodes (%s): %s", name, err, buf.String())
			return err
		}
	}
	return nil
}

// newContainer creates a new container in Docker and stores it in the database.
func newContainer(app provision.App, imageId string, cmds []string) (container, error) {
	cont := container{
		AppName: app.GetName(),
		Type:    app.GetPlatform(),
	}
	port, err := getPort()
	if err != nil {
		log.Errorf("error on getting port for container %s - %s", cont.AppName, port)
		return container{}, err
	}
	user, _ := config.GetString("docker:ssh:user")
	config := docker.Config{
		Image:        imageId,
		Cmd:          cmds,
		User:         user,
		PortSpecs:    []string{port},
		AttachStdin:  false,
		AttachStdout: false,
		AttachStderr: false,
	}
	hostID, c, err := dockerCluster().CreateContainer(&config)
	if err != nil {
		log.Errorf("error on creating container in docker %s - %s", cont.AppName, err)
		return container{}, err
	}
	cont.ID = c.ID
	cont.Port = port
	cont.HostAddr = getHostAddr(hostID)
	return cont, nil
}

func handle(msg *queue.Message) {
	if msg.Action == addUnitToLoadBalancer {
		if len(msg.Args) < 1 {
			log.Errorf("Failed to handle %q: it requires at least one argument.", msg.Action)
			return
		}
		a := qApp{name: msg.Args[0]}
		unitNames := msg.Args[1:]
		sort.Strings(unitNames)
		status, err := (&JujuProvisioner{}).collectStatus()
		if err != nil {
			log.Errorf("Failed to handle %q: juju status failed.\n%s.", msg.Action, err)
			return
		}
		var units []provision.Unit
		for _, u := range status {
			if u.AppName != a.name {
				continue
			}
			n := sort.SearchStrings(unitNames, u.Name)
			if len(unitNames) == 0 ||
				n < len(unitNames) && unitNames[n] == u.Name {
				units = append(units, u)
			}
		}
		if len(units) == 0 {
			log.Errorf("Failed to handle %q: units not found.", msg.Action)
			return
		}
		var noID []string
		var ok []provision.Unit
		for _, u := range units {
			if u.InstanceId == "pending" || u.InstanceId == "" {
				noID = append(noID, u.Name)
			} else {
				ok = append(ok, u)
			}
		}
		if len(noID) == len(units) {
			getQueue(queueName).Put(msg, 0)
		} else {
			router, _ := Router()
			for _, u := range units {
				router.AddRoute(a.GetName(), u.InstanceId)
			}
			if len(noID) > 0 {
				args := []string{a.name}
				args = append(args, noID...)
				msg := queue.Message{
					Action: msg.Action,
					Args:   args,
				}
				getQueue(queueName).Put(&msg, 1e9)
			}
		}
	}
}

func removeContainer(c *container) error {
	err := c.stop()
	if err != nil {
		log.Errorf("error on stop unit %s - %s", c.ID, err)
	}
	err = c.remove()
	if err != nil {
		log.Errorf("error on remove container %s - %s", c.ID, err)
	}
	return err
}

// newContainer creates a new container in Docker and stores it in the database.
func newContainer(app provision.App, imageId string, cmds []string) (container, error) {
	contName := containerName()
	cont := container{
		AppName: app.GetName(),
		Type:    app.GetPlatform(),
		Name:    contName,
		Status:  "created",
	}
	coll := collection()
	defer coll.Close()
	if err := coll.Insert(cont); err != nil {
		log.Errorf("error on inserting container into database %s - %s", cont.Name, err)
		return container{}, err
	}
	port, err := getPort()
	if err != nil {
		log.Errorf("error on getting port for container %s - %s", cont.AppName, port)
		return container{}, err
	}
	user, _ := config.GetString("docker:ssh:user")
	exposedPorts := make(map[docker.Port]struct{}, 1)
	p := docker.Port(fmt.Sprintf("%s/tcp", port))
	exposedPorts[p] = struct{}{}
	config := docker.Config{
		Image:        imageId,
		Cmd:          cmds,
		User:         user,
		ExposedPorts: exposedPorts,
		AttachStdin:  false,
		AttachStdout: false,
		AttachStderr: false,
	}
	opts := dclient.CreateContainerOptions{Name: contName}
	hostID, c, err := dockerCluster().CreateContainer(opts, &config)
	if err == dclient.ErrNoSuchImage {
		var buf bytes.Buffer
		pullOpts := dclient.PullImageOptions{Repository: imageId}
		dockerCluster().PullImage(pullOpts, &buf)
		hostID, c, err = dockerCluster().CreateContainer(opts, &config)
	}
	if err != nil {
		log.Errorf("error on creating container in docker %s - %s", cont.AppName, err)
		return container{}, err
	}
	cont.ID = c.ID
	cont.Port = port
	cont.HostAddr = getHostAddr(hostID)
	err = coll.Update(bson.M{"name": cont.Name}, cont)
	if err != nil {
		log.Errorf("error on updating container into database %s - %s", cont.ID, err)
		return container{}, err
	}
	return cont, nil
}

func (u *User) sendResetPassword(t *passwordToken) {
	var body bytes.Buffer
	err := resetEmailData.Execute(&body, t)
	if err != nil {
		log.Errorf("Failed to send password token to user %q: %s", u.Email, err)
		return
	}
	err = sendEmail(u.Email, body.Bytes())
	if err != nil {
		log.Errorf("Failed to send password token for user %q: %s", u.Email, err)
	}
}

func (*dockerProvisioner) Addr(app provision.App) (string, error) {
	r, err := getRouter()
	if err != nil {
		log.Errorf("Failed to get router: %s", err)
		return "", err
	}
	addr, err := r.Addr(app.GetName())
	if err != nil {
		log.Errorf("Failed to obtain app %s address: %s", app.GetName(), err)
		return "", err
	}
	return addr, nil
}

func injectEnvsAndRestart(a provision.App) {
	time.Sleep(5e9)
	err := a.SerializeEnvVars()
	if err != nil {
		log.Errorf("Failed to serialize env vars: %s.", err)
	}
	var buf bytes.Buffer
	w := app.LogWriter{App: a, Writer: &buf}
	err = a.Restart(&w)
	if err != nil {
		log.Errorf("Failed to restart app %q (%s): %s.", a.GetName(), err, buf.String())
	}
}

// Creates a new user and write his/her keys into authorized_keys file.
//
// The authorized_keys file belongs to the user running the process.
func New(name string, keys map[string]string) (*User, error) {
	log.Debugf(`Creating user "%s"`, name)
	u := &User{Name: name}
	if v, err := u.isValid(); !v {
		log.Errorf("user.New: %s", err.Error())
		return u, err
	}
	if err := db.Session.User().Insert(&u); err != nil {
		log.Errorf("user.New: %s", err.Error())
		return u, err
	}
	return u, addKeys(keys, u.Name)
}

// removeUser removes the user from the database and from gandalf server
//
// If the user is the only one in a team an error will be returned.
func removeUser(w http.ResponseWriter, r *http.Request, t *auth.Token) error {
	u, err := t.User()
	if err != nil {
		return err
	}
	gURL := repository.ServerURL()
	c := gandalf.Client{Endpoint: gURL}
	alwdApps, err := u.AllowedApps()
	if err != nil {
		return err
	}
	if err := c.RevokeAccess(alwdApps, []string{u.Email}); err != nil {
		log.Errorf("Failed to revoke access in Gandalf: %s", err)
		return fmt.Errorf("Failed to revoke acess from git repositories: %s", err)
	}
	teams, err := u.Teams()
	if err != nil {
		return err
	}
	conn, err := db.Conn()
	if err != nil {
		return err
	}
	defer conn.Close()
	for _, team := range teams {
		if len(team.Users) < 2 {
			msg := fmt.Sprintf(`This user is the last member of the team "%s", so it cannot be removed.

Please remove the team, them remove the user.`, team.Name)
			return &errors.HTTP{Code: http.StatusForbidden, Message: msg}
		}
		err = team.RemoveUser(u)
		if err != nil {
			return err
		}
		// this can be done without the loop
		err = conn.Teams().Update(bson.M{"_id": team.Name}, team)
		if err != nil {
			return err
		}
	}
	rec.Log(u.Email, "remove-user")
	if err := c.RemoveUser(u.Email); err != nil {
		log.Errorf("Failed to remove user from gandalf: %s", err)
		return fmt.Errorf("Failed to remove the user from the git server: %s", err)
	}
	quota.Delete(u.Email)
	return conn.Users().Remove(bson.M{"email": u.Email})
}

func (p *dockerProvisioner) Stop(app provision.App) error {
	containers, err := listAppContainers(app.GetName())
	if err != nil {
		log.Errorf("Got error while getting app containers: %s", err)
		return nil
	}
	for _, c := range containers {
		err := c.stop()
		if err != nil {
			log.Errorf("Failed to stop %q: %s", app.GetName(), err)
			return err
		}
	}
	return nil
}

func setQueue() {
	var err error
	qfactory, err = queue.Factory()
	if err != nil {
		log.Errorf("Failed to get the queue instance: %s", err)
	}
	_handler, err = qfactory.Handler(handle, queueName)
	if err != nil {
		log.Errorf("Failed to create the queue handler: %s", err)
	}
	_queue, err = qfactory.Get(queueName)
	if err != nil {
		log.Errorf("Failed to get the queue instance: %s", err)
	}
}

// handle is the function called by the queue handler on each message.
func handle(msg *queue.Message) {
	switch msg.Action {
	case RegenerateApprcAndStart:
		fallthrough
	case regenerateApprc:
		if len(msg.Args) < 1 {
			log.Errorf("Error handling %q: this action requires at least 1 argument.", msg.Action)
			msg.Delete()
			return
		}
		app, err := ensureAppIsStarted(msg)
		if err != nil {
			log.Error(err.Error())
			return
		}
		msg.Delete()
		app.SerializeEnvVars()
		fallthrough
	case startApp:
		if msg.Action == regenerateApprc {
			break
		}
		if len(msg.Args) < 1 {
			log.Errorf("Error handling %q: this action requires at least 1 argument.", msg.Action)
		}
		app, err := ensureAppIsStarted(msg)
		if err != nil {
			log.Error(err.Error())
			return
		}
		err = app.Restart(ioutil.Discard)
		if err != nil {
			log.Errorf("Error handling %q. App failed to start:\n%s.", msg.Action, err)
			return
		}
		msg.Delete()
	case BindService:
		err := bindUnit(msg)
		if err != nil {
			log.Error(err.Error())
			return
		}
		msg.Delete()
	default:
		log.Errorf("Error handling %q: invalid action.", msg.Action)
		msg.Delete()
	}
}

func (u *User) sendNewPassword(password string) {
	m := map[string]string{
		"password": password,
		"email":    u.Email,
	}
	var body bytes.Buffer
	err := passwordResetConfirm.Execute(&body, m)
	if err != nil {
		log.Errorf("Failed to send new password to user %q: %s", u.Email, err)
		return
	}
	err = sendEmail(u.Email, body.Bytes())
	if err != nil {
		log.Errorf("Failed to send new password to user %q: %s", u.Email, err)
	}
}

func (p *dockerProvisioner) Destroy(app provision.App) error {
	containers, _ := listAppContainers(app.GetName())
	go func(c []container) {
		var containersGroup sync.WaitGroup
		containersGroup.Add(len(containers))
		for _, c := range containers {
			go func(c container) {
				defer containersGroup.Done()
				err := removeContainer(&c)
				if err != nil {
					log.Error(err.Error())
				}
			}(c)
		}
		containersGroup.Wait()
		err := removeImage(assembleImageName(app.GetName()))
		if err != nil {
			log.Error(err.Error())
		}
	}(containers)
	r, err := getRouter()
	if err != nil {
		log.Errorf("Failed to get router: %s", err)
		return err
	}
	return r.RemoveBackend(app.GetName())
}

func collectUnit(container container, units chan<- provision.Unit, wg *sync.WaitGroup) {
	defer wg.Done()
	unit := provision.Unit{
		Name:    container.ID,
		AppName: container.AppName,
		Type:    container.Type,
	}
	if container.Status == "error" {
		unit.Status = provision.StatusDown
		units <- unit
		return
	}
	if container.Status == "running" {
		unit.Ip = container.HostAddr
		if ip, hostPort, err := container.networkInfo(); err == nil &&
			(hostPort != container.HostPort || ip != container.IP) {
			err = fixContainer(&container, ip, hostPort)
			if err != nil {
				log.Errorf("error on fix container hostport for [container %s]", container.ID)
				return
			}
		}
		addr := strings.Replace(container.getAddress(), "http://", "", 1)
		conn, err := net.Dial("tcp", addr)
		if err != nil {
			unit.Status = provision.StatusUnreachable
		} else {
			conn.Close()
			unit.Status = provision.StatusStarted
		}
		log.Debugf("collected data for [container %s] - [app %s]", container.ID, container.AppName)
		units <- unit
	}
}