// TestProtoEncodingError verifies that MarshalResponse and
// UnmarshalRequest gracefully handles a protocol message type error.
func TestProtoEncodingError(t *testing.T) {
	req, err := http.NewRequest("GET", "http://foo.com", nil)
	if err != nil {
		t.Fatal(err)
	}
	req.Header.Add(util.ContentTypeHeader, util.ProtoContentType)
	reqBody := []byte("foo")
	var value string
	err = util.UnmarshalRequest(req, reqBody, value, []util.EncodingType{util.ProtoEncoding})
	if err == nil {
		t.Errorf("unexpected success")
	}

	req.Header.Add(util.AcceptHeader, util.ProtoContentType)
	body, cType, err := util.MarshalResponse(req, value, []util.EncodingType{util.ProtoEncoding})
	if err != nil {
		t.Fatal(err)
	}
	if cType != util.JSONContentType {
		t.Errorf("unexpected content type; got %s", cType)
	}
	if !bytes.Equal(body, body) {
		t.Errorf("unexpected boy; got %s", body)
	}
}

// ServeHTTP serves the SQL API by treating the request URL path
// as the method, the request body as the arguments, and sets the
// response body as the method reply. The request body is unmarshalled
// into arguments based on the Content-Type request header. Protobuf
// and JSON-encoded requests are supported. The response body is
// encoded according to the request's Accept header, or if not
// present, in the same format as the request's incoming Content-Type
// header.
func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	defer r.Body.Close()
	method := r.URL.Path
	if !strings.HasPrefix(method, driver.Endpoint) {
		http.Error(w, http.StatusText(http.StatusNotFound), http.StatusNotFound)
		return
	}

	// Check TLS settings.
	authenticationHook, err := security.AuthenticationHook(s.context.Insecure, r.TLS)
	if err != nil {
		http.Error(w, err.Error(), http.StatusUnauthorized)
		return
	}

	method = strings.TrimPrefix(method, driver.Endpoint)
	if method != driver.Execute.String() {
		http.Error(w, http.StatusText(http.StatusNotFound), http.StatusNotFound)
		return
	}

	// Unmarshal the request.
	reqBody, err := ioutil.ReadAll(r.Body)
	if err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}

	var args driver.Request
	if err := util.UnmarshalRequest(r, reqBody, &args, allowedEncodings); err != nil {
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}

	// Check request user against client certificate user.
	if err := authenticationHook(&args); err != nil {
		http.Error(w, err.Error(), http.StatusUnauthorized)
		return
	}

	// Send the Request for SQL execution and set the application-level error
	// on the reply.
	reply, err := s.exec(args)
	if err != nil {
		errProto := proto.Error{}
		errProto.SetResponseGoError(err)
		reply.Error = &errProto
	}

	// Marshal the response.
	body, contentType, err := util.MarshalResponse(r, &reply, allowedEncodings)
	if err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}
	w.Header().Set(util.ContentTypeHeader, contentType)
	w.Write(body)
}

// ServeHTTP serves the SQL API by treating the request URL path
// as the method, the request body as the arguments, and sets the
// response body as the method reply. The request body is unmarshalled
// into arguments based on the Content-Type request header. Protobuf
// and JSON-encoded requests are supported. The response body is
// encoded according to the request's Accept header, or if not
// present, in the same format as the request's incoming Content-Type
// header.
func (s Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	defer r.Body.Close()
	method := r.URL.Path
	if !strings.HasPrefix(method, driver.Endpoint) {
		http.Error(w, http.StatusText(http.StatusNotFound), http.StatusNotFound)
		return
	}

	// Check TLS settings.
	authenticationHook, err := security.ProtoAuthHook(s.context.Insecure, r.TLS)
	if err != nil {
		http.Error(w, err.Error(), http.StatusUnauthorized)
		return
	}

	method = strings.TrimPrefix(method, driver.Endpoint)
	if method != driver.Execute.String() {
		http.Error(w, http.StatusText(http.StatusNotFound), http.StatusNotFound)
		return
	}

	// Unmarshal the request.
	reqBody, err := ioutil.ReadAll(r.Body)
	if err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}

	var args driver.Request
	if err := util.UnmarshalRequest(r, reqBody, &args, allowedEncodings); err != nil {
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}

	// Check request user against client certificate user.
	if err := authenticationHook(&args, true /* public */); err != nil {
		http.Error(w, err.Error(), http.StatusUnauthorized)
		return
	}

	reply, code, err := s.Execute(args)
	if err != nil {
		http.Error(w, err.Error(), code)
	}

	// Marshal the response.
	body, contentType, err := util.MarshalResponse(r, &reply, allowedEncodings)
	if err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}
	w.Header().Set(util.ContentTypeHeader, contentType)
	if _, err := w.Write(body); err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
	}
}

// ServeHTTP serves the key-value API by treating the request URL path
// as the method, the request body as the arguments, and sets the
// response body as the method reply. The request body is unmarshalled
// into arguments based on the Content-Type request header. Protobuf
// and JSON-encoded requests are supported. The response body is
// encoded according the the request's Accept header, or if not
// present, in the same format as the request's incoming Content-Type
// header.
func (s *DBServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	// Check TLS settings before anything else.
	authenticationHook, err := security.AuthenticationHook(s.context.Insecure, r.TLS)
	if err != nil {
		http.Error(w, err.Error(), http.StatusUnauthorized)
		return
	}

	method := r.URL.Path
	if !strings.HasPrefix(method, DBPrefix) {
		http.Error(w, http.StatusText(http.StatusNotFound), http.StatusNotFound)
		return
	}
	method = strings.TrimPrefix(method, DBPrefix)
	args, reply := createArgsAndReply(method)
	if args == nil {
		http.Error(w, http.StatusText(http.StatusNotFound), http.StatusNotFound)
		return
	}

	// Unmarshal the request.
	reqBody, err := ioutil.ReadAll(r.Body)
	defer r.Body.Close()
	if err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}
	if err := util.UnmarshalRequest(r, reqBody, args, allowedEncodings); err != nil {
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}

	// Verify the request for public API.
	if err := verifyRequest(args); err != nil {
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}

	// Check request user against client certificate user.
	if err := authenticationHook(args); err != nil {
		http.Error(w, err.Error(), http.StatusUnauthorized)
		return
	}

	// Create a call and invoke through sender.
	s.sender.Send(context.TODO(), proto.Call{Args: args, Reply: reply})

	// Marshal the response.
	body, contentType, err := util.MarshalResponse(r, reply, allowedEncodings)
	if err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}
	w.Header().Set(util.ContentTypeHeader, contentType)
	w.Write(body)
}

// TestHTTPSenderSend verifies sending posts.
func TestHTTPSenderSend(t *testing.T) {
	defer leaktest.AfterTest(t)
	server, addr := startTestHTTPServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		// Make sure SSL certs were properly specified.
		authenticationHook, err := security.AuthenticationHook(false /* !insecure */, r.TLS)
		if err != nil {
			t.Error(err)
		}

		if r.Method != "POST" {
			t.Errorf("expected method POST; got %s", r.Method)
		}
		if r.URL.Path != KVDBEndpoint+"Put" {
			t.Errorf("expected url %s; got %s", KVDBEndpoint+"Put", r.URL.Path)
		}
		// Unmarshal the request.
		reqBody, err := ioutil.ReadAll(r.Body)
		if err != nil {
			t.Errorf("unexpected error reading body: %s", err)
		}

		args := &proto.PutRequest{}
		if err := util.UnmarshalRequest(r, reqBody, args, util.AllEncodings); err != nil {
			t.Errorf("unexpected error unmarshalling request: %s", err)
		}

		// Validate request against incoming user.
		if err := authenticationHook(args, false /*not public*/); err != nil {
			t.Error(err)
		}

		if !args.Key.Equal(testPutReq.Key) || !args.Timestamp.Equal(testPutReq.Timestamp) {
			t.Errorf("expected parsed %+v to equal %+v", args, testPutReq)
		}
		body, contentType, err := util.MarshalResponse(r, testPutResp, util.AllEncodings)
		if err != nil {
			t.Errorf("failed to marshal response: %s", err)
		}
		w.Header().Set(util.ContentTypeHeader, contentType)
		w.Write(body)
	}))
	defer server.Close()

	sender, err := newHTTPSender(addr, testutils.NewNodeTestBaseContext(), defaultRetryOptions)
	if err != nil {
		t.Fatal(err)
	}
	reply := &proto.PutResponse{}
	sender.Send(context.Background(), proto.Call{Args: testPutReq, Reply: reply})
	if reply.GoError() != nil {
		t.Errorf("expected success; got %s", reply.GoError())
	}
	if !reply.Timestamp.Equal(testPutResp.Timestamp) {
		t.Errorf("expected received %+v to equal %+v", reply, testPutResp)
	}
}

// handleQuery handles an incoming HTTP query request. Each query requests data
// for one or more metrics over a specific time span. Query requests have a
// significant body and thus are POST requests.
func (s *Server) handleQuery(w http.ResponseWriter, r *http.Request, _ httprouter.Params) {
	request := &TimeSeriesQueryRequest{}

	// Unmarshal query request.
	reqBody, err := ioutil.ReadAll(r.Body)
	if err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}
	if err := util.UnmarshalRequest(r, reqBody, request, util.AllEncodings); err != nil {
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}

	if len(request.Queries) == 0 {
		http.Error(w, "time series query requests must specify at least one query.", http.StatusBadRequest)
		return
	}

	response := &TimeSeriesQueryResponse{
		Results: make([]*TimeSeriesQueryResponse_Result, 0, len(request.Queries)),
	}
	for _, q := range request.Queries {
		datapoints, sources, err := s.db.Query(q, Resolution10s, request.StartNanos, request.EndNanos)
		if err != nil {
			http.Error(w, err.Error(), http.StatusInternalServerError)
			return
		}
		result := &TimeSeriesQueryResponse_Result{
			Query:      q,
			Datapoints: datapoints,
		}
		// TODO(tamird): Remove this (and all other) explicit setting of defaults.
		// It is currently required because the client side doesn't know about
		// proto defaults.
		result.SourceAggregator = q.GetSourceAggregator().Enum()
		result.Downsampler = q.GetDownsampler().Enum()
		result.Derivative = q.GetDerivative().Enum()

		result.Sources = sources
		response.Results = append(response.Results, result)
	}

	// Marshal and return response.
	b, contentType, err := util.MarshalResponse(r, response, util.AllEncodings)
	if err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}
	w.Header().Set(util.ContentTypeHeader, contentType)
	if _, err := w.Write(b); err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}
}

// ServeHTTP serves the key-value API by treating the request URL path
// as the method, the request body as the arguments, and sets the
// response body as the method reply. The request body is unmarshalled
// into arguments based on the Content-Type request header. Protobuf
// and JSON-encoded requests are supported. The response body is
// encoded according the the request's Accept header, or if not
// present, in the same format as the request's incoming Content-Type
// header.
func (s *DBServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	method := r.URL.Path
	if !strings.HasPrefix(method, DBPrefix) {
		http.Error(w, http.StatusText(http.StatusNotFound), http.StatusNotFound)
		return
	}
	method = strings.TrimPrefix(method, DBPrefix)
	if !proto.IsPublic(method) {
		http.Error(w, http.StatusText(http.StatusNotFound), http.StatusNotFound)
		return
	}

	// Unmarshal the request.
	reqBody, err := ioutil.ReadAll(r.Body)
	defer r.Body.Close()
	if err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}
	args, reply, err := proto.CreateArgsAndReply(method)
	if err != nil {
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}
	if err := util.UnmarshalRequest(r, reqBody, args, allowedEncodings); err != nil {
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}

	// Verify the request for public API.
	if err := verifyRequest(args); err != nil {
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}

	// Create a call and invoke through sender.
	call := &client.Call{
		Method: method,
		Args:   args,
		Reply:  reply,
	}
	s.sender.Send(call)

	// Marshal the response.
	body, contentType, err := util.MarshalResponse(r, reply, allowedEncodings)
	if err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}
	w.Header().Set("Content-Type", contentType)
	w.Write(body)
}

// Put writes a perm config for the specified key prefix (which is treated as
// a key). The perm config is parsed from the input "body". The perm config is
// stored gob-encoded. The specified body must validly parse into a
// perm config struct.
func (ph *permHandler) Put(path string, body []byte, r *http.Request) error {
	if len(path) == 0 {
		return util.Errorf("no path specified for permission Put")
	}
	config := &proto.PermConfig{}
	if err := util.UnmarshalRequest(r, body, config, util.AllEncodings); err != nil {
		return util.Errorf("permission config has invalid format: %s: %s", config, err)
	}
	permKey := engine.MakeKey(engine.KeyConfigPermissionPrefix, proto.Key(path[1:]))
	if err := ph.db.PutProto(permKey, config); err != nil {
		return err
	}
	return nil
}

// handleQuery handles an incoming HTTP query request. Each query requests data
// for one or more metrics over a specific time span. Query requests have a
// significant body and thus are POST requests.
func (s *Server) handleQuery(w http.ResponseWriter, r *http.Request, _ httprouter.Params) {
	request := &TimeSeriesQueryRequest{}

	// Unmarshal query request.
	reqBody, err := ioutil.ReadAll(r.Body)
	defer r.Body.Close()
	if err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}
	if err := util.UnmarshalRequest(r, reqBody, request, util.AllEncodings); err != nil {
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}

	if len(request.Queries) == 0 {
		http.Error(w, "time series query requests must specify at least one query.", http.StatusBadRequest)
		return
	}

	response := &TimeSeriesQueryResponse{
		Results: make([]*TimeSeriesQueryResponse_Result, 0, len(request.Queries)),
	}
	for _, q := range request.Queries {
		datapoints, sources, err := s.db.Query(q, Resolution10s, request.StartNanos, request.EndNanos)
		if err != nil {
			http.Error(w, err.Error(), http.StatusInternalServerError)
			return
		}
		response.Results = append(response.Results, &TimeSeriesQueryResponse_Result{
			Name:       q.Name,
			Sources:    sources,
			Datapoints: datapoints,
			Aggregator: q.Aggregator,
		})
	}

	// Marshal and return response.
	b, contentType, err := util.MarshalResponse(r, response, util.AllEncodings)
	if err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}
	w.Header().Set(util.ContentTypeHeader, contentType)
	if _, err := w.Write(b); err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}
}

// putConfig writes a config for the specified key prefix (which is
// treated as a key). The config is parsed from the input "body". The
// config is stored proto-encoded. The specified body must validly
// parse into a config struct and must pass a given validation check (if
// validate is not nil).
func putConfig(db *client.DB, configPrefix proto.Key, config gogoproto.Message,
	path string, body []byte, r *http.Request,
	validate func(gogoproto.Message) error) error {
	if len(path) == 0 {
		return util.Errorf("no path specified for Put")
	}
	if err := util.UnmarshalRequest(r, body, config, util.AllEncodings); err != nil {
		return util.Errorf("config has invalid format: %+v: %s", config, err)
	}
	if validate != nil {
		if err := validate(config); err != nil {
			return err
		}
	}
	key := keys.MakeKey(configPrefix, proto.Key(path[1:]))
	return db.Put(key, config)
}

// ServeHTTP serves the SQL API by treating the request URL path
// as the method, the request body as the arguments, and sets the
// response body as the method reply. The request body is unmarshalled
// into arguments based on the Content-Type request header. Protobuf
// and JSON-encoded requests are supported. The response body is
// encoded according to the request's Accept header, or if not
// present, in the same format as the request's incoming Content-Type
// header.
func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	method := r.URL.Path
	if !strings.HasPrefix(method, sqlwire.Endpoint) {
		http.Error(w, http.StatusText(http.StatusNotFound), http.StatusNotFound)
		return
	}
	method = strings.TrimPrefix(method, sqlwire.Endpoint)
	args, reply := createArgsAndReply(method)
	if args == nil {
		http.Error(w, http.StatusText(http.StatusNotFound), http.StatusNotFound)
		return
	}

	// Unmarshal the request.
	reqBody, err := ioutil.ReadAll(r.Body)
	defer r.Body.Close()
	if err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}
	if err := util.UnmarshalRequest(r, reqBody, args, allowedEncodings); err != nil {
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}

	// Send the SQLRequest for SQL execution.
	if httpStatus, err := s.execute(args, reply); err != nil {
		http.Error(w, err.Error(), httpStatus)
		return
	}

	// Marshal the response.
	body, contentType, err := util.MarshalResponse(r, reply, allowedEncodings)
	if err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}
	w.Header().Set(util.ContentTypeHeader, contentType)
	w.Write(body)
}

func TestUnmarshalRequest(t *testing.T) {
	testCases := []struct {
		cType    string
		body     []byte
		expError bool
	}{
		{util.JSONContentType, jsonConfig, false},
		{util.AltJSONContentType, jsonConfig, false},
		{util.ProtoContentType, protobufConfig, false},
		{util.AltProtoContentType, protobufConfig, false},
		{util.YAMLContentType, yamlConfig, false},
		{util.AltYAMLContentType, yamlConfig, false},
		{"foo", jsonConfig, true},
		{"baz", protobufConfig, true},
		{"bar", yamlConfig, true},
	}

	for i, test := range testCases {
		req, err := http.NewRequest("GET", "http://foo.com", nil)
		if err != nil {
			t.Fatal(err)
		}
		req.Header.Add(util.ContentTypeHeader, test.cType)
		config := &proto.ZoneConfig{}
		err = util.UnmarshalRequest(req, test.body, config, util.AllEncodings)
		if test.expError {
			if err == nil {
				t.Errorf("%d: unexpected success", i)
			}
			continue
		} else if !test.expError && err != nil {
			t.Errorf("%d: unexpected failure: %s", i, err)
			continue
		}
		if !reflect.DeepEqual(config, &testConfig) {
			t.Errorf("%d: unmarshalling yielded config %+v; expected %+v", i, config, testConfig)
		}
	}
}

// ServeHTTP serves the SQL API by treating the request URL path
// as the method, the request body as the arguments, and sets the
// response body as the method reply. The request body is unmarshalled
// into arguments based on the Content-Type request header. Protobuf
// and JSON-encoded requests are supported. The response body is
// encoded according to the request's Accept header, or if not
// present, in the same format as the request's incoming Content-Type
// header.
func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	defer r.Body.Close()
	method := r.URL.Path
	if !strings.HasPrefix(method, driver.Endpoint) {
		http.Error(w, http.StatusText(http.StatusNotFound), http.StatusNotFound)
		return
	}

	// Check TLS settings.
	authenticationHook, err := security.AuthenticationHook(s.context.Insecure, r.TLS)
	if err != nil {
		http.Error(w, err.Error(), http.StatusUnauthorized)
		return
	}

	method = strings.TrimPrefix(method, driver.Endpoint)
	if method != driver.Execute.String() {
		http.Error(w, http.StatusText(http.StatusNotFound), http.StatusNotFound)
		return
	}

	// Unmarshal the request.
	reqBody, err := ioutil.ReadAll(r.Body)
	if err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}

	var args driver.Request
	if err := util.UnmarshalRequest(r, reqBody, &args, allowedEncodings); err != nil {
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}

	// Check request user against client certificate user.
	if err := authenticationHook(&args, true /*public*/); err != nil {
		http.Error(w, err.Error(), http.StatusUnauthorized)
		return
	}

	// Pick up current session state.
	planMaker := planner{user: args.GetUser()}
	if err := gogoproto.Unmarshal(args.Session, &planMaker.session); err != nil {
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}
	// Open a pending transaction if needed.
	if planMaker.session.Txn != nil {
		txn := client.NewTxn(*s.db)
		txn.Proto = *planMaker.session.Txn
		planMaker.txn = txn
	}

	// Send the Request for SQL execution and set the application-level error
	// for each result in the reply.
	reply := s.exec(args, &planMaker)

	// Send back the session state even if there were application-level errors.
	// Add transaction to session state.
	if planMaker.txn != nil {
		planMaker.session.Txn = &planMaker.txn.Proto
	} else {
		planMaker.session.Txn = nil
	}
	bytes, err := gogoproto.Marshal(&planMaker.session)
	if err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}
	reply.Session = bytes

	// Marshal the response.
	body, contentType, err := util.MarshalResponse(r, &reply, allowedEncodings)
	if err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}
	w.Header().Set(util.ContentTypeHeader, contentType)
	w.Write(body)
}