// Subscribe watches etcd changes and generates structured events telling vulcand to add or delete frontends, hosts etc.
// It is a blocking function.
func (n *ng) Subscribe(changes chan interface{}, cancelC chan bool) error {
	// This index helps us to get changes in sequence, as they were performed by clients.
	waitIndex := uint64(0)
	for {
		response, err := n.client.Watch(n.etcdKey, waitIndex, true, nil, cancelC)
		if err != nil {
			switch err {
			case etcd.ErrWatchStoppedByUser:
				log.Infof("Stop watching: graceful shutdown")
				return nil
			default:
				log.Errorf("unexpected error: %s, stop watching", err)
				return err
			}
		}
		waitIndex = response.EtcdIndex + 1
		log.Infof("%s", responseToString(response))
		change, err := n.parseChange(response)
		if err != nil {
			log.Warningf("Ignore '%s', error: %s", responseToString(response), err)
			continue
		}
		if change != nil {
			log.Infof("%v", change)
			select {
			case changes <- change:
			case <-cancelC:
				return nil
			}
		}
	}
}

// Start the app on the configured host/port.
//
// Supports graceful shutdown on 'kill' and 'int' signals.
func (app *App) Run() error {
	if app.vulcandReg != nil {
		err := app.vulcandReg.Start()
		if err != nil {
			return fmt.Errorf("failed to start vulcand registry: err=(%s)", err)
		}
		heartbeatCh := make(chan os.Signal, 1)
		signal.Notify(heartbeatCh, syscall.SIGUSR1)
		go func() {
			sig := <-heartbeatCh
			log.Infof("Got signal %v, canceling vulcand registration", sig)
			app.vulcandReg.Stop()
		}()
	}

	// listen for a shutdown signal
	exitCh := make(chan os.Signal, 1)
	signal.Notify(exitCh, syscall.SIGINT, syscall.SIGQUIT, syscall.SIGTERM)
	go func() {
		s := <-exitCh
		log.Infof("Got signal %v, shutting down", s)
		if app.vulcandReg != nil {
			app.vulcandReg.Stop()
		}
		manners.Close()
	}()

	addr := fmt.Sprintf("%v:%v", app.Config.ListenIP, app.Config.ListenPort)
	return manners.ListenAndServe(addr, app.router)
}

// Start the app on the configured host/port.
//
// Supports graceful shutdown on 'kill' and 'int' signals.
func (app *App) Run() error {
	// toggle heartbeat on SIGUSR1
	go func() {
		app.heartbeater.Start()
		heartbeatChan := make(chan os.Signal, 1)
		signal.Notify(heartbeatChan, syscall.SIGUSR1)

		for s := range heartbeatChan {
			log.Infof("Received signal: %v, toggling heartbeat", s)
			app.heartbeater.Toggle()
		}
	}()

	// listen for a shutdown signal
	go func() {
		exitChan := make(chan os.Signal, 1)
		signal.Notify(exitChan, syscall.SIGINT, syscall.SIGQUIT, syscall.SIGTERM)
		s := <-exitChan
		log.Infof("Got shutdown signal: %v", s)
		manners.Close()
	}()

	addr := fmt.Sprintf("%v:%v", app.Config.ListenIP, app.Config.ListenPort)
	return manners.ListenAndServe(addr, app.router)
}

func proxyHandler(w http.ResponseWriter, r *http.Request) {

	vars := mux.Vars(r)
	domain := vars["domain"]
	reqPath := vars["path"]
	if domain == "github.com" {
		domain = "raw.githubusercontent.com"
		reqPath = strings.Replace(reqPath, "/blob", "", 1)
	}

	url := fmt.Sprintf("%s/%s", domain, reqPath)
	cachePath := fmt.Sprintf("%s/%s", tmpDir, url)

	// Check for file in cache.
	if useFileCache {
		contents, err := ioutil.ReadFile(path.Clean(cachePath))
		if err == nil {

			log.Infof("HIT %s", cachePath)
			addCacheHeader(&w, "yes")
			w.Header().Set("Content-Type", mime.TypeByExtension(path.Ext(reqPath)))
			w.Write(contents)

			return
		}
	}

	resp, err := http.Get(fmt.Sprintf("http://%s", url))
	if err != nil {
		log.Infof("ERROR %s", err.Error())
		http.Error(w, http.StatusText(500), 500)
		return
	}

	if resp.StatusCode >= 400 {
		http.Error(w, resp.Status, resp.StatusCode)
		return
	}

	log.Infof("PROXY %s", url)
	w.Header().Set("Content-Type", r.Header.Get("Content-Type"))

	if useFileCache {
		writeAndCache(w, resp.Body, cachePath)
		return
	}

	io.Copy(w, resp.Body)
	return

}

func (r *ratioController) allowRequest() bool {
	log.Infof("%v", r)
	t := r.targetRatio()
	// This condition answers the question - would we satisfy the target ratio if we allow this request?
	e := r.computeRatio(r.allowed+1, r.denied)
	if e < t {
		r.allowed++
		log.Infof("%v allowed", r)
		return true
	}
	r.denied++
	log.Infof("%v denied", r)
	return false
}

// RegisterApp adds a new backend and a single server with Vulcand.
func (s *LeaderRegistry) RegisterApp(registration *AppRegistration) error {
	log.Infof("Registering app: %v", registration)

	endpoint, err := vulcan.NewEndpointWithID(s.Group, registration.Name, registration.Host, registration.Port)
	if err != nil {
		return err
	}

	err = s.client.RegisterBackend(endpoint)
	if err != nil {
		log.Errorf("Failed to register backend for endpoint: %v, %s", endpoint, err)
		return err
	}

	if s.IsMaster {
		err = s.maintainLeader(endpoint)
	} else {
		err = s.initLeader(endpoint)
	}

	if err != nil {
		log.Errorf("Failed to register server for endpoint: %v, %s", endpoint, err)
		return err
	}

	return nil
}

func (w *WebhookSideEffect) Exec() error {
	r, err := http.NewRequest(w.w.Method, w.w.URL, w.getBody())
	if err != nil {
		return err
	}
	if len(w.w.Headers) != 0 {
		utils.CopyHeaders(r.Header, w.w.Headers)
	}
	if len(w.w.Form) != 0 {
		r.Header.Set("Content-Type", "application/x-www-form-urlencoded")
	}
	re, err := http.DefaultClient.Do(r)
	if err != nil {
		return err
	}
	if re.Body != nil {
		defer re.Body.Close()
	}
	body, err := ioutil.ReadAll(re.Body)
	if err != nil {
		return err
	}
	log.Infof("%v got response: (%s): %s", w, re.Status, string(body))
	return nil
}

// Stop halts sending heartbeats.
func (h *Heartbeater) Stop() {
	log.Infof("Stopping heartbeat for app: %v", h.registration)

	close(h.quit)
	h.ticker.Stop()
	h.Running = false
}

// Make a handler out of HandlerWithBodyFunc, just like regular MakeHandler function.
func MakeHandlerWithBody(app *App, fn HandlerWithBodyFunc, spec Spec) http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {
		if err := parseForm(r); err != nil {
			ReplyInternalError(w, fmt.Sprintf("Failed to parse request form: %v", err))
			return
		}

		body, err := ioutil.ReadAll(r.Body)
		if err != nil {
			ReplyInternalError(w, fmt.Sprintf("Failed to read request body: %v", err))
			return
		}

		start := time.Now()
		response, err := fn(w, r, mux.Vars(r), body)
		elapsedTime := time.Since(start)

		var status int
		if err != nil {
			response, status = responseAndStatusFor(err)
		} else {
			status = http.StatusOK
		}

		log.Infof("Request(Status=%v, Method=%v, Path=%v, Form=%v, Time=%v, Error=%v)",
			status, r.Method, r.URL, r.Form, elapsedTime, err)

		app.stats.TrackRequest(spec.MetricName, status, elapsedTime)

		Reply(w, response, status)
	}
}

func (c *ProxyController) deleteFrontend(w http.ResponseWriter, r *http.Request, params map[string]string) (interface{}, error) {
	log.Infof("Delete Frontend(id=%s)", params["id"])
	if err := c.ng.DeleteFrontend(engine.FrontendKey{Id: params["id"]}); err != nil {
		return nil, formatError(err)
	}
	return scroll.Response{"message": "Frontend deleted"}, nil
}

func (c *ProxyController) upsertFrontend(w http.ResponseWriter, r *http.Request, params map[string]string, body []byte) (interface{}, error) {
	frontend, ttl, err := parseFrontendPack(c.ng.GetRegistry().GetRouter(), body)
	if err != nil {
		return nil, formatError(err)
	}
	log.Infof("Upsert %s", frontend)
	return formatResult(frontend, c.ng.UpsertFrontend(*frontend, ttl))
}

func (c *ProxyController) upsertBackend(w http.ResponseWriter, r *http.Request, params map[string]string, body []byte) (interface{}, error) {
	b, err := parseBackendPack(body)
	if err != nil {
		return nil, formatError(err)
	}
	log.Infof("Upsert Backend: %s", b)
	return formatResult(b, c.ng.UpsertBackend(*b))
}

func (c *ProxyController) deleteHost(w http.ResponseWriter, r *http.Request, params map[string]string) (interface{}, error) {
	hostname := params["hostname"]
	log.Infof("Delete host: %s", hostname)
	if err := c.ng.DeleteHost(engine.HostKey{Name: hostname}); err != nil {
		return nil, formatError(err)
	}
	return scroll.Response{"message": fmt.Sprintf("Host '%s' deleted", hostname)}, nil
}

func (c *ProxyController) upsertHost(w http.ResponseWriter, r *http.Request, params map[string]string, body []byte) (interface{}, error) {
	host, err := parseHostPack(body)
	if err != nil {
		return nil, formatError(err)
	}
	log.Infof("Upsert %s", host)
	return formatResult(host, c.ng.UpsertHost(*host))
}

func (c *ProxyController) deleteServer(w http.ResponseWriter, r *http.Request, params map[string]string) (interface{}, error) {
	sk := engine.ServerKey{BackendKey: engine.BackendKey{Id: params["backendId"]}, Id: params["id"]}
	log.Infof("Delete %v", sk)
	if err := c.ng.DeleteServer(sk); err != nil {
		return nil, formatError(err)
	}
	return scroll.Response{"message": "Server deleted"}, nil
}

func initCacheDir() {
	if useFileCache {
		if err := os.MkdirAll(tmpDir, os.FileMode(0775)); err != nil {
			useFileCache = false
			log.Errorf("Couldn't create tmp dir %s: %s", tmpDir, err)
		} else {
			if err := ioutil.WriteFile(fmt.Sprintf("%s/lock", tmpDir), []byte(version), 0664); err != nil {
				useFileCache = false
				log.Errorf("Couldn't write to tmp dir %s: %s", tmpDir, err)
			}
		}
		if useFileCache {
			log.Infof("Caching via filesystem enabled")
			log.Infof("Using %v as cache path", tmpDir)
		}
	}
}

func (c *ProxyController) upsertListener(w http.ResponseWriter, r *http.Request, params map[string]string, body []byte) (interface{}, error) {
	listener, err := parseListenerPack(body)
	if err != nil {
		return nil, formatError(err)
	}
	log.Infof("Upsert %s", listener)
	return formatResult(listener, c.ng.UpsertListener(*listener))
}

func (c *ProxyController) getServer(w http.ResponseWriter, r *http.Request, params map[string]string) (interface{}, error) {
	sk := engine.ServerKey{BackendKey: engine.BackendKey{Id: params["backendId"]}, Id: params["id"]}
	log.Infof("getServer %v", sk)
	srv, err := c.ng.GetServer(sk)
	if err != nil {
		return nil, formatError(err)
	}
	return formatResult(srv, err)
}

// Start begins sending heartbeats.
func (h *Heartbeater) Start() {
	log.Infof("Starting heartbeat for app: %v", h.registration)

	h.Running = true
	h.ticker = time.NewTicker(h.interval)
	h.quit = make(chan int)

	go h.heartbeat()
}

func (s *LeaderRegistry) maintainLeader(endpoint *vulcan.Endpoint) error {
	err := s.client.UpdateServer(endpoint, s.TTL)
	if err != nil {
		log.Infof("Falling back to follow role for endpoint: %v", endpoint)
		s.IsMaster = false
		return err
	}

	return nil
}