// RegisterHTTPForever is just like RegisterHTTP however it first tries to
// register forever until a response from kontrol is received. It's useful to
// use it during app initializations. After the registration a reconnect is
// automatically handled inside the RegisterHTTP method.
func (k *Kite) RegisterHTTPForever(kiteURL *url.URL) {
	// Create the httpBackoffRegister that RegisterHTTPForever will
	// use to backoff repeated register attempts.
	httpRegisterBackOff := backoff.NewExponentialBackOff()
	httpRegisterBackOff.InitialInterval = 30 * time.Second
	httpRegisterBackOff.MaxInterval = 5 * time.Minute
	httpRegisterBackOff.Multiplier = 1.7
	httpRegisterBackOff.MaxElapsedTime = 0

	register := func() error {
		_, err := k.RegisterHTTP(kiteURL)
		if err != nil {
			k.Log.Error("Cannot register to Kontrol: %s Will retry after %d seconds",
				err,
				httpRegisterBackOff.NextBackOff()/time.Second)
			return err
		}

		return nil
	}

	// this will retry register forever
	err := backoff.Retry(register, httpRegisterBackOff)
	if err != nil {
		k.Log.Error("BackOff stopped retrying with Error '%s'", err)
	}
}

func refreshFromSource() {
	certData := bytes.NewBuffer([]byte{})

	bo := backoff.NewExponentialBackOff()
	bo.MaxElapsedTime = time.Minute

	err := backoff.Retry(func() error {
		src, err := http.Get(certDataSource)
		if err != nil {
			return err
		}
		defer src.Body.Close()

		license, cvsID, objects := cert.ParseInput(src.Body)

		fmt.Fprint(certData, license)
		if len(cvsID) > 0 {
			fmt.Fprint(certData, "CVS_ID "+cvsID+"\n")
		}

		cert.OutputTrustedCerts(certData, objects)

		return nil
	}, bo)

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

	saveToCache(strconv.FormatInt(time.Now().UTC().Unix(), 10), certData.Bytes())
	latestCertificates = certData.Bytes()
}

func (as *APISelector) makeImageRequest(urlString string, bodyLines []string) (*apiSelectorImageResponse, error) {
	var responseBody []byte

	b := backoff.NewExponentialBackOff()
	b.MaxInterval = 10 * time.Second
	b.MaxElapsedTime = time.Minute

	err := backoff.Retry(func() (err error) {
		resp, err := http.Post(urlString, imageAPIRequestContentType,
			strings.NewReader(strings.Join(bodyLines, "\n")+"\n"))

		if err != nil {
			return err
		}
		defer resp.Body.Close()
		responseBody, err = ioutil.ReadAll(resp.Body)
		return
	}, b)

	if err != nil {
		return nil, err
	}

	imageResp := &apiSelectorImageResponse{
		Data: []*apiSelectorImageRef{},
	}

	err = json.Unmarshal(responseBody, imageResp)
	if err != nil {
		return nil, err
	}

	return imageResp, nil
}

func (c *Client) makeRequest(request *batch) {
	payload, err := json.Marshal(request)
	if err != nil {
		log.Printf("[Error] Batch failed to marshal: %v - %v", request, err)
		return
	}

	bodyReader := bytes.NewReader(payload)

	b := backoff.NewExponentialBackOff()
	b.MaxElapsedTime = 10 * time.Second
	err = backoff.Retry(func() error {
		resp, err := http.Post(c.BaseEndpoint+"/v1/set", "application/json", bodyReader)
		if err != nil {
			return err
		}
		defer resp.Body.Close()

		response := map[string]interface{}{}
		dec := json.NewDecoder(resp.Body)
		dec.Decode(&response)

		if resp.StatusCode != http.StatusOK {
			return fmt.Errorf("HTTP Post Request Failed, Status Code %d: %v", resp.StatusCode, response)
		}

		return nil
	}, b)

	if err != nil {
		log.Printf("[Error] %v", err)
		return
	}
}

func (i *gceInstance) isPreempted(ctx gocontext.Context) (bool, error) {
	if !i.ic.Preemptible {
		return false, nil
	}

	listOpCall := i.provider.client.GlobalOperations.AggregatedList(i.provider.projectID).
		Filter(fmt.Sprintf("targetId eq %d", i.instance.Id))

	b := backoff.NewExponentialBackOff()
	b.InitialInterval = 1 * time.Second
	b.MaxElapsedTime = 1 * time.Minute

	var preempted bool
	err := backoff.Retry(func() error {
		i.provider.apiRateLimit(ctx)
		list, err := listOpCall.Do()
		if err != nil {
			return err
		}

		for _, item := range list.Items {
			for _, op := range item.Operations {
				if op.Kind == "compute#operation" && op.OperationType == "compute.instances.preempted" {
					preempted = true
					return nil
				}
			}
		}

		return nil
	}, b)

	return preempted, err
}

func (s *stepGenerateScript) Run(state multistep.StateBag) multistep.StepAction {
	buildJob := state.Get("buildJob").(Job)
	ctx := state.Get("ctx").(gocontext.Context)

	b := backoff.NewExponentialBackOff()
	b.MaxInterval = 10 * time.Second
	b.MaxElapsedTime = time.Minute

	var script []byte
	err := backoff.Retry(func() (err error) {
		script, err = s.generator.Generate(ctx, buildJob.RawPayload())
		return
	}, b)

	if err != nil {
		context.LoggerFromContext(ctx).WithField("err", err).Error("couldn't generate build script, erroring job")
		err := buildJob.Error(ctx, "An error occurred while generating the build script.")
		if err != nil {
			context.LoggerFromContext(ctx).WithField("err", err).Error("couldn't requeue job")
		}

		return multistep.ActionHalt
	}

	context.LoggerFromContext(ctx).Info("generated script")

	state.Put("script", script)

	return multistep.ActionContinue
}

func makeJobBoardImagesRequest(urlString string) (*jobBoardImagesResponse, error) {
	var responseBody []byte

	b := backoff.NewExponentialBackOff()
	b.MaxInterval = 10 * time.Second
	b.MaxElapsedTime = time.Minute

	err := backoff.Retry(func() (err error) {
		resp, err := http.Get(urlString)

		if err != nil {
			return err
		}
		defer resp.Body.Close()
		responseBody, err = ioutil.ReadAll(resp.Body)
		return
	}, b)

	if err != nil {
		return nil, err
	}

	imageResp := &jobBoardImagesResponse{
		Data: []*jobBoardImageRef{},
	}

	err = json.Unmarshal(responseBody, imageResp)
	if err != nil {
		return nil, err
	}

	return imageResp, nil
}

func retryTask(c context.Context, ds appwrap.Datastore, taskIntf TaskInterface, jobKey *datastore.Key, taskKey *datastore.Key) error {
	var job JobInfo

	if j, err := getJob(ds, jobKey); err != nil {
		return fmt.Errorf("getting job: %s", err)
	} else {
		job = j
	}

	time.Sleep(time.Duration(job.RetryCount) * 5 * time.Second)

	if err := backoff.Retry(func() error {
		var task JobTask
		if err := ds.Get(taskKey, &task); err != nil {
			return fmt.Errorf("getting task: %s", err)
		}

		task.Status = TaskStatusPending
		if _, err := ds.Put(taskKey, &task); err != nil {
			return fmt.Errorf("putting task: %s", err)
		} else if err := taskIntf.PostTask(c, task.Url, job.JsonParameters); err != nil {
			return fmt.Errorf("enqueuing task: %s", err)
		}

		logInfo(c, "retrying task %d/%d", task.Retries, job.RetryCount)
		return nil
	}, mrBackOff()); err != nil {
		logInfo(c, "retryTask() failed after backoff attempts")
		return err
	} else {
		return nil
	}
}

// listenForNodeChanges listens for changes to node status using change feeds.
// This function will block until the query fails
func (c *Cluster) listenForNodeChanges() error {
	// Start listening to changes from a random active node
	node, hpr, err := c.GetNextNode()
	if err != nil {
		return err
	}

	q, err := newQuery(
		DB("rethinkdb").Table("server_status").Changes(),
		map[string]interface{}{},
		c.opts,
	)
	if err != nil {
		return fmt.Errorf("Error building query: %s", err)
	}

	cursor, err := node.Query(q)
	if err != nil {
		hpr.Mark(err)
		return err
	}

	// Keep reading node status updates from changefeed
	var result struct {
		NewVal nodeStatus `gorethink:"new_val"`
		OldVal nodeStatus `gorethink:"old_val"`
	}
	for cursor.Next(&result) {
		addr := fmt.Sprintf("%s:%d", result.NewVal.Network.Hostname, result.NewVal.Network.ReqlPort)
		addr = strings.ToLower(addr)

		switch result.NewVal.Status {
		case "connected":
			// Connect to node using exponential backoff (give up after waiting 5s)
			// to give the node time to start-up.
			b := backoff.NewExponentialBackOff()
			b.MaxElapsedTime = time.Second * 5

			backoff.Retry(func() error {
				node, err := c.connectNodeWithStatus(result.NewVal)
				if err == nil {
					if !c.nodeExists(node) {
						c.addNode(node)

						Log.WithFields(logrus.Fields{
							"id":   node.ID,
							"host": node.Host.String(),
						}).Debug("Connected to node")
					}
				}

				return err
			}, b)
		}
	}

	err = cursor.Err()
	hpr.Mark(err)
	return err
}

func watch(src string) error {
	loc, err := storage.ParseLocation(src)
	if err != nil {
		return err
	}

	watcher, err := sync.Watch(loc)
	if err != nil {
		return err
	}
	defer watcher.Stop()

	// TODO: a better approach here would be to use a channel to retry on,
	//       then if you jacked up the config, it would pick up the change
	//       in the middle of all the retries. As it stands now it would take a
	//       minute to fix itself.
	eb := backoff.NewExponentialBackOff()
	eb.MaxElapsedTime = time.Minute

	for range watcher.C {
		log.Println("[watch] new version")
		backoff.Retry(func() error {
			err := apply(src)
			if err != nil {
				log.Printf("[watch] error installing: %v\n", err)
			}
			return err
		}, eb)
	}

	return nil
}

func main() {
	flag.Parse()
	args := flag.Args()
	if len(args) == 0 {
		usage()
	}

	var b []byte
	operation := func() error {
		var err error
		b, err = exec.Command(flag.Arg(0), args[1:]...).Output()
		if err != nil {
			log.Printf("err: %s", err)
		}
		return err
	}

	bf := backoff.NewExponentialBackOff()
	second := func(i int) time.Duration {
		return time.Duration(i) * time.Second
	}

	bf.MaxElapsedTime = second(*flagMaxElapsedTime)
	bf.MaxInterval = second(*flagMaxInterval)
	bf.InitialInterval = second(*flagInitialInterval)

	err := backoff.Retry(operation, bf)
	if err != nil {
		fmt.Fprintf(os.Stderr, "operation failed: %s\n", err)
		os.Exit(1)
	}

	fmt.Fprint(os.Stdout, string(b))
	os.Exit(0)
}

func (b *Bridge) Add(containerId string) {
	b.Lock()
	defer b.Unlock()
	container, err := b.docker.InspectContainer(containerId)
	if err != nil {
		log.Println("docksul: unable to inspect container:", containerId, err)
		return
	}

	portDefs := make([][]string, 0)
	for port, published := range container.NetworkSettings.Ports {
		if len(published) > 0 {
			p := strings.Split(string(port), "/")
			portDefs = append(portDefs, []string{published[0].HostPort, p[0], p[1]})
		}
	}

	multiservice := len(portDefs) > 1
	for _, port := range portDefs {
		service := b.buildService(container, port[0], port[1], port[2], multiservice)
		err := backoff.Retry(func() error {
			return b.consul.Agent().ServiceRegister(service)
		}, backoff.NewExponentialBackoff())
		if err != nil {
			log.Println("docksul: unable to register service:", service, err)
			continue
		}
		b.services[container.ID] = append(b.services[container.ID], service)
		log.Println("docksul: added:", container.ID[:12], service.ID)
	}
}

func (s *Sender) connect() {

	for s.conn == nil {
		var conn *conn
		var err error

		connect := func() error {
			log.Printf("Connecting to %v", s.addr)
			conn, err = newConn(s.addr, s.cert)
			if err != nil {
				log.Printf("Failed connecting to %v: %v; will retry", s.addr, err)
				return err
			}
			return nil
		}

		if backoff.Retry(connect, backoff.NewExponentialBackOff()) != nil {
			continue
		}

		log.Printf("Connected to %v", s.addr)

		go s.read(conn)

		s.conn = conn
	}
}

// main reads configs, starts all gouroutines and waits until a message is in channel stop.
func main() {
	backoff.Retry(testWebGuiPost, backoff.NewExponentialBackOff())
	// Attempt to increase the limit on number of open files to the maximum allowed.
	MaximizeOpenFileLimit()

	allFolders := getFolders()
	folders := filterFolders(allFolders)
	if len(folders) == 0 {
		log.Fatalln("No folders to be watched, exiting...")
	}
	stChans := make(map[string]chan STEvent, len(folders))
	for _, folder := range folders {
		Debug.Println("Installing watch for " + folder.Label)
		stChan := make(chan STEvent)
		stChans[folder.ID] = stChan
		go watchFolder(folder, stChan)
	}
	// Note: Lose thread ownership of stChans
	go watchSTEvents(stChans, allFolders)
	go listenForSighup()

	code := <-stop
	OK.Println("Exiting")
	os.Exit(code)
}

// createUnitWithRetry wraps CreateUnit of the fleet API with a retry.
func (f *FleetTunnel) createUnitWithRetry(unit *schema.Unit) error {
	op := func() error {
		return maskAny(f.cAPI.CreateUnit(unit))
	}
	if err := backoff.Retry(op, backoff.NewExponentialBackOff()); err != nil {
		return maskAny(err)
	}
	return nil
}

// setUnitTargetStateWithRetry wraps SetUnitTargetState of the fleet API with a retry.
func (f *FleetTunnel) setUnitTargetStateWithRetry(name, target string) error {
	op := func() error {
		return maskAny(f.cAPI.SetUnitTargetState(name, target))
	}
	if err := backoff.Retry(op, backoff.NewExponentialBackOff()); err != nil {
		return maskAny(err)
	}
	return nil
}

func (p proxy) ServeHTTP(res http.ResponseWriter, r *http.Request) {
	bo := backoff.NewExponentialBackOff()
	bo.MaxElapsedTime = 5 * time.Second
	if err := backoff.Retry(func() error {
		r.URL.Host = base.Host
		r.URL.Scheme = base.Scheme
		r.RequestURI = ""
		r.Host = base.Host

		suppliedToken := r.URL.Query().Get("token")
		if authCookie, err := r.Cookie("grafana-proxy-auth"); err == nil {
			suppliedToken = authCookie.Value
		}

		if cfg.Token != "" && suppliedToken != cfg.Token {
			http.Error(res, "Please add the `?token=xyz` parameter with correct token", http.StatusForbidden)
			return nil
		}

		resp, err := client.Do(r)
		if err != nil {
			return err
		}

		defer resp.Body.Close()

		res.Header().Del("Content-Type")
		for k, v := range resp.Header {
			for _, v1 := range v {
				res.Header().Add(k, v1)
			}
		}

		if r.URL.Query().Get("token") != "" {
			http.SetCookie(res, &http.Cookie{
				Name:   "grafana-proxy-auth",
				Value:  r.URL.Query().Get("token"),
				MaxAge: 31536000, // 1 Year
				Path:   "/",
			})
		}

		if resp.StatusCode == 401 {
			loadLogin()
			return fmt.Errorf("Need to relogin")
		}

		res.WriteHeader(resp.StatusCode)
		written, _ := io.Copy(res, resp.Body)

		log.Printf("%s %s?%s %d %d\n", r.Method, r.URL.Path, r.URL.RawQuery, resp.StatusCode, written)
		return nil
	}, bo); err != nil {
		http.Error(res, fmt.Sprintf("Woot?\n%s", err), http.StatusInternalServerError)
	}
}

// retryingFindTXT will, on any DNS failure, retry for up to 15 minutes before
// giving up and returning an empty []string of records
func retryingFindTXT(fqdn string) (records []string, ttl time.Duration, err error) {
	err = backoff.Retry(
		func() error {
			records, ttl, err = findTXT(fqdn)
			if err != nil {
				log.Error("Retrying DNS query. Query failed with: %s", err.Error())
			}
			return err
		}, backoff.NewExponentialBackOff())
	return
}

func createTasks(ds appwrap.Datastore, jobKey *datastore.Key, taskKeys []*datastore.Key, tasks []JobTask, newStage JobStage) error {
	now := time.Now()
	firstId := taskKeys[0].IntID()
	for i := range tasks {
		tasks[i].StartTime = now
		tasks[i].Job = jobKey

		if taskKeys[i].IntID() < firstId {
			firstId = taskKeys[i].IntID()
		}
	}

	putSize := 64

	i := 0
	for i < len(tasks) {
		if err := backoff.Retry(func() error {
			last := i + putSize
			if last > len(tasks) {
				last = len(tasks)
			}

			if _, err := ds.PutMulti(taskKeys[i:last], tasks[i:last]); err != nil {
				if putSize > 5 {
					putSize /= 2
				}

				return err
			}

			i = last

			return nil
		}, mrBackOff()); err != nil {
			return err
		}
	}

	return runInTransaction(ds,
		func(ds appwrap.Datastore) error {
			var job JobInfo

			if err := ds.Get(jobKey, &job); err != nil {
				return err
			}

			job.TaskCount = len(tasks)
			job.FirstTaskId = firstId
			job.Stage = newStage

			_, err := ds.Put(jobKey, &job)
			return err
		})
}

func startServiceContainer(tenantName string) error {
	// do nothing in test mode
	if testMode {
		return nil
	}

	var err error
	docker, err := utils.GetDockerClient()
	if err != nil {
		log.Errorf("Unable to connect to docker. Error %v", err)
		return err
	}

	// pull the skydns image if it does not exist
	imageName := defaultSkyDNSImage
	_, err = docker.InspectImage(imageName)
	if err != nil {
		pullOperation := func() error {
			err := docker.PullImage(imageName, nil)
			if err != nil {
				log.Errorf("Retrying to pull image: %s", imageName)
				return err
			}
			return nil
		}

		err = backoff.Retry(pullOperation, backoff.NewExponentialBackOff())
		if err != nil {
			log.Errorf("Unable to pull image: %s", imageName)
			return err
		}
	}

	containerConfig := &dockerclient.ContainerConfig{
		Image: imageName,
		Env: []string{"ETCD_MACHINES=http://172.17.0.1:4001",
			"SKYDNS_NAMESERVERS=8.8.8.8:53",
			"SKYDNS_ADDR=0.0.0.0:53",
			"SKYDNS_DOMAIN=" + tenantName}}

	containerID, err := docker.CreateContainer(containerConfig, tenantName+"dns")
	if err != nil {
		log.Errorf("Error creating DNS container for tenant: %s. Error: %s", tenantName, err)
	}

	// Start the container
	err = docker.StartContainer(containerID, nil)
	if err != nil {
		log.Errorf("Error starting DNS container for tenant: %s. Error: %s", tenantName, err)
	}

	return err
}