// 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 getFileFromUpstream(fileURL string, useSSL bool) (body []byte, statusCode int, err error) {

	// Try to fetch the image. If not, fail.
	url := fileURL
	if useSSL {
		url = fmt.Sprintf("https://%s", url)
	} else {
		url = fmt.Sprintf("http://%s", url)
	}

	resp, e := http.Get(url)
	if e != nil {
		err = e
		log.Errorf("Couldn't get %s", url)
		statusCode = 500
		return
	}
	defer resp.Body.Close()

	// Make sure the request succeeded
	if resp.StatusCode > 302 {
		log.Errorf("Couldn't fetch %s", url)
		statusCode = resp.StatusCode
		return
	}

	body, err = ioutil.ReadAll(resp.Body)
	if err != nil {
		log.Errorf("Couldn't read body %s", url)
		statusCode = 500
	}

	return

}

// Round trips the request to one of the endpoints and returns the response.
func (l *HttpLocation) RoundTrip(req request.Request) (*http.Response, error) {
	// Get options and transport as one single read transaction.
	// Options and transport may change if someone calls SetOptions
	o, tr := l.GetOptionsAndTransport()
	originalRequest := req.GetHttpRequest()

	//  Check request size first, if that exceeds the limit, we don't bother reading the request.
	if l.isRequestOverLimit(req) {
		return nil, errors.FromStatus(http.StatusRequestEntityTooLarge)
	}

	// Read the body while keeping this location's limits in mind. This reader controls the maximum bytes
	// to read into memory and disk. This reader returns anerror if the total request size exceeds the
	// prefefined MaxSizeBytes. This can occur if we got chunked request, in this case ContentLength would be set to -1
	// and the reader would be unbounded bufio in the http.Server
	body, err := netutils.NewBodyBufferWithOptions(originalRequest.Body, netutils.BodyBufferOptions{
		MemBufferBytes: o.Limits.MaxMemBodyBytes,
		MaxSizeBytes:   o.Limits.MaxBodyBytes,
	})
	if err != nil {
		return nil, err
	}
	if body == nil {
		return nil, fmt.Errorf("Empty body")
	}

	// Set request body to buffered reader that can replay the read and execute Seek
	req.SetBody(body)
	// Note that we don't change the original request Body as it's handled by the http server
	defer body.Close()

	for {
		_, err := req.GetBody().Seek(0, 0)
		if err != nil {
			return nil, err
		}

		endpoint, err := l.loadBalancer.NextEndpoint(req)
		if err != nil {
			log.Errorf("Load Balancer failure: %s", err)
			return nil, err
		}

		// Adds headers, changes urls. Note that we rewrite request each time we proxy it to the
		// endpoint, so that each try gets a fresh start
		req.SetHttpRequest(l.copyRequest(originalRequest, req.GetBody(), endpoint))

		// In case if error is not nil, we allow load balancer to choose the next endpoint
		// e.g. to do request failover. Nil error means that we got proxied the request successfully.
		response, err := l.proxyToEndpoint(tr, &o, endpoint, req)
		if o.ShouldFailover(req) {
			continue
		} else {
			return response, err
		}
	}
	log.Errorf("All endpoints failed!")
	return nil, fmt.Errorf("All endpoints failed")
}

func main() {
	log.InitWithConfig(log.Config{Name: "console"})

	r, err := registry.GetRegistry()
	if err != nil {
		log.Errorf("Error: %s\n", err)
		return
	}
	cmd := command.NewCommand(r)
	if err := cmd.Run(os.Args); err != nil {
		log.Errorf("Error: %s\n", err)
	}
}

// Proxy the request to the given endpoint, execute observers and middlewares chains
func (l *HttpLocation) proxyToEndpoint(tr *http.Transport, o *Options, endpoint endpoint.Endpoint, req request.Request) (*http.Response, error) {

	a := &request.BaseAttempt{Endpoint: endpoint}

	l.observerChain.ObserveRequest(req)
	defer l.observerChain.ObserveResponse(req, a)
	defer req.AddAttempt(a)

	it := l.middlewareChain.GetIter()
	defer l.unwindIter(it, req, a)

	for v := it.Next(); v != nil; v = it.Next() {
		a.Response, a.Error = v.ProcessRequest(req)
		if a.Response != nil || a.Error != nil {
			// Move the iterator forward to count it again once we unwind the chain
			it.Next()
			log.Errorf("Midleware intercepted request with response=%s, error=%s", a.Response.Status, a.Error)
			return a.Response, a.Error
		}
	}

	// Forward the request and mirror the response
	start := o.TimeProvider.UtcNow()
	a.Response, a.Error = tr.RoundTrip(req.GetHttpRequest())
	a.Duration = o.TimeProvider.UtcNow().Sub(start)
	return a.Response, a.Error
}

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

// Round trips the request to the selected location and writes back the response
func (p *Proxy) proxyRequest(w http.ResponseWriter, r *http.Request) error {

	// Create a unique request with sequential ids that will be passed to all interfaces.
	req := request.NewBaseRequest(r, atomic.AddInt64(&p.lastRequestId, 1), nil)
	location, err := p.router.Route(req)
	if err != nil {
		return err
	}

	// Router could not find a matching location, we can do nothing else.
	if location == nil {
		log.Errorf("%s failed to route", req)
		return errors.FromStatus(http.StatusBadGateway)
	}

	response, err := location.RoundTrip(req)
	if response != nil {
		netutils.CopyHeaders(w.Header(), response.Header)
		w.WriteHeader(response.StatusCode)
		io.Copy(w, response.Body)
		response.Body.Close()
		return nil
	} else {
		return err
	}
}

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)
		}
	}
}

// registerLocationForScope registers a location with a specified scope.
func (app *App) registerLocationForScope(methods []string, path string, scope Scope, middlewares []middleware.Middleware) {
	host, err := app.apiHostForScope(scope)
	if err != nil {
		log.Errorf("Failed to register a location: %v", err)
		return
	}
	app.registerLocationForHost(methods, path, host, middlewares)
}

// RegisterHandler registers the frontends and middlewares with Vulcand.
func (s *LBRegistry) RegisterHandler(registration *HandlerRegistration) error {
	log.Infof("Registering handler: %v", registration)

	location := vulcan.NewLocation(registration.Host, registration.Methods, registration.Path, registration.Name, registration.Middlewares)
	err := s.client.RegisterFrontend(location)
	if err != nil {
		log.Errorf("Failed to register frontend for location: %v, %s", location, err)
		return err
	}

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

	return nil
}

func (n *ng) newHttpTransport() etcd.CancelableTransport {

	var cc *tls.Config = nil

	if n.options.EtcdCertFile != "" && n.options.EtcdKeyFile != "" {
		var rpool *x509.CertPool = nil
		if n.options.EtcdCaFile != "" {
			if pemBytes, err := ioutil.ReadFile(n.options.EtcdCaFile); err == nil {
				rpool = x509.NewCertPool()
				rpool.AppendCertsFromPEM(pemBytes)
			} else {
				log.Errorf("Error reading Etcd Cert CA File: %v", err)
			}
		}

		if tlsCert, err := tls.LoadX509KeyPair(n.options.EtcdCertFile, n.options.EtcdKeyFile); err == nil {
			cc = &tls.Config{
				RootCAs:            rpool,
				Certificates:       []tls.Certificate{tlsCert},
				InsecureSkipVerify: true,
			}
		} else {
			log.Errorf("Error loading KeyPair for TLS client: %v", err)
		}

	}

	//Copied from etcd.DefaultTransport declaration
	//Wasn't sure how to make a clean reliable deep-copy, and instead
	//creating a new object was the safest and most reliable assurance
	//that we aren't overwriting some global struct potentially
	//shared by other etcd users.
	tr := &http.Transport{
		Proxy: http.ProxyFromEnvironment,
		Dial: (&net.Dialer{
			Timeout:   30 * time.Second,
			KeepAlive: 30 * time.Second,
		}).Dial,
		TLSHandshakeTimeout: 10 * time.Second,
		TLSClientConfig:     cc,
	}

	return tr
}

func (rw *rewriteHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
	oldURL := rawURL(req)

	// only continue if the Regexp param matches the URL
	if !rw.regexp.MatchString(oldURL) {
		rw.next.ServeHTTP(w, req)
		return
	}

	// apply a rewrite regexp to the URL
	newURL := rw.regexp.ReplaceAllString(oldURL, rw.replacement)

	// replace any variables that may be in there
	rewrittenURL := &bytes.Buffer{}
	if err := ApplyString(newURL, rewrittenURL, req); err != nil {
		rw.errHandler.ServeHTTP(w, req, err)
		return
	}

	// parse the rewritten URL and replace request URL with it
	parsedURL, err := url.Parse(rewrittenURL.String())
	if err != nil {
		rw.errHandler.ServeHTTP(w, req, err)
		return
	}

	if rw.redirect && newURL != oldURL {
		(&redirectHandler{u: parsedURL}).ServeHTTP(w, req)
		return
	}

	req.URL = parsedURL

	// make sure the request URI corresponds the rewritten URL
	req.RequestURI = req.URL.RequestURI()

	if !rw.rewriteBody {
		rw.next.ServeHTTP(w, req)
		return
	}

	bw := &bufferWriter{header: make(http.Header), buffer: &bytes.Buffer{}}
	newBody := &bytes.Buffer{}

	rw.next.ServeHTTP(bw, req)

	if err := Apply(bw.buffer, newBody, req); err != nil {
		log.Errorf("Failed to rewrite response body: %v", err)
		return
	}

	utils.CopyHeaders(w.Header(), bw.Header())
	w.Header().Set("Content-Length", strconv.Itoa(newBody.Len()))
	w.WriteHeader(bw.code)
	io.Copy(w, newBody)
}

func (f *JsonFormatter) Format(err ProxyError) (int, []byte, string) {
	encodedError, e := json.Marshal(map[string]interface{}{
		"error": string(err.Error()),
	})
	if e != nil {
		log.Errorf("Failed to serialize: %s", e)
		encodedError = []byte("{}")
	}
	return err.GetStatusCode(), encodedError, "application/json"
}

// exec executes side effect
func (c *CircuitBreaker) exec(s SideEffect) {
	if s == nil {
		return
	}
	go func() {
		if err := s.Exec(); err != nil {
			log.Errorf("%v side effect failure: %v", c, err)
		}
	}()
}

// cacheOptimizedImageWriter Returns a file and a writer for the image resource.
// Be careful to flush the writer and close the file manually, as this function
// doesn't do that.
func fileCacheWriter(fileName string) (f *os.File, w *bufio.Writer) {

	dir := path.Dir(fileName)
	err := os.MkdirAll(dir, os.FileMode(0775))
	if err != nil {
		log.Errorf("Error caching image %s: %s", fileName, err)
		return
	}

	f, err = os.Create(fileName)
	if err != nil {
		log.Errorf("Error caching image %s: %s", fileName, err)
		return
	}

	w = bufio.NewWriter(f)

	return

}

func cacheFile(fileName string, data []byte) {
	if useFileCache && len(data) > 0 {

		go func(fileName string, data []byte) {

			fileName = path.Clean(fileName)
			dir := path.Dir(fileName)

			if err := os.MkdirAll(dir, os.FileMode(0775)); err != nil {
				log.Errorf("Error caching image %s: %s", fileName, err)
				return
			}

			if err := ioutil.WriteFile(fileName, data, 0644); err != nil {
				log.Errorf("Error caching image %s: %s", fileName, err)
				return
			}

		}(fileName, data)

	}
}

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

	endpoint, err := vulcan.NewEndpoint(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
	}

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

	return nil
}

func latencyAtQuantile(quantile float64) threshold.RequestToInt {
	return func(r request.Request) int {
		m := getMetrics(r)
		if m == nil {
			return 0
		}
		h, err := m.GetLatencyHistogram()
		if err != nil {
			log.Errorf("Failed to get latency histogram, for %v error: %v", r, err)
			return 0
		}
		return int(h.LatencyAtQuantile(quantile) / time.Millisecond)
	}
}

func (r *Registry) heartbeat() {
	defer r.wg.Done()
	tick := time.NewTicker(r.cfg.TTL * 3 / 4)
	for {
		select {
		case <-tick.C:
			if err := r.registerBackendServer(r.backendSpec, r.cfg.TTL); err != nil {
				log.Errorf("Heartbeat failed: err=(%v)", err)
			}
		case <-r.ctx.Done():
			err := r.removeBackendServer(r.backendSpec)
			log.Infof("Heartbeat stopped: err=(%v)", err)
			return
		}
	}
}

func (m *RoundTripMetrics) recordStatusCode(a request.Attempt) {
	if a.GetResponse() == nil {
		return
	}
	statusCode := a.GetResponse().StatusCode
	if c, ok := m.statusCodes[statusCode]; ok {
		c.Inc()
		return
	}
	c, err := m.newCounter()
	if err != nil {
		log.Errorf("failed to create a counter: %v", err)
		return
	}
	c.Inc()
	m.statusCodes[statusCode] = c
}