Add endpoint for updating event type

event/errors.go

@@ -29,10 +29,10 @@ func (e ErrEventTypeValidation) Error() string {
 	return fmt.Sprintf("Event Type doesn't validate. Validation error: %s", e.Message)
 }
 
-// ErrEventTypeHasSubscriptionsError occurs when there are subscription for the event type.
-type ErrEventTypeHasSubscriptionsError struct{}
+// ErrEventTypeHasSubscriptions occurs when there are subscription for the event type.
+type ErrEventTypeHasSubscriptions struct{}
 
-func (e ErrEventTypeHasSubscriptionsError) Error() string {
+func (e ErrEventTypeHasSubscriptions) Error() string {
 	return fmt.Sprintf("Event type cannot be deleted because there are subscriptions using it.")
 }
 

event/service.go

@@ -5,5 +5,6 @@ type Service interface {
 	CreateEventType(eventType *Type) (*Type, error)
 	GetEventType(space string, name TypeName) (*Type, error)
 	GetEventTypes(space string) (Types, error)
+	UpdateEventType(newEventType *Type) (*Type, error)
 	DeleteEventType(space string, name TypeName) error
 }

function/errors.go

@@ -67,9 +67,9 @@ func (e ErrFunctionError) Error() string {
 	return fmt.Sprintf("Function call failed because of runtime error. Error: %s", e.Original)
 }
 
-// ErrFunctionHasSubscriptionsError occurs when function with subscription is being deleted.
-type ErrFunctionHasSubscriptionsError struct{}
+// ErrFunctionHasSubscriptions occurs when function with subscription is being deleted.
+type ErrFunctionHasSubscriptions struct{}
 
-func (e ErrFunctionHasSubscriptionsError) Error() string {
+func (e ErrFunctionHasSubscriptions) Error() string {
 	return fmt.Sprintf("Function cannot be deleted because it's subscribed to a least one event.")
 }

httpapi/httpapi.go

@@ -41,6 +41,7 @@ func (h HTTPAPI) RegisterRoutes(router *httprouter.Router) {
 	router.GET("/v1/spaces/:space/eventtypes", h.listEventTypes)
 	router.GET("/v1/spaces/:space/eventtypes/:name", h.getEventType)
 	router.POST("/v1/spaces/:space/eventtypes", h.createEventType)
+	router.PUT("/v1/spaces/:space/eventtypes/:name", h.updateEventType)
 	router.DELETE("/v1/spaces/:space/eventtypes/:name", h.deleteEventType)
 
 	router.GET("/v1/spaces/:space/functions", h.listFunctions)
@@ -129,6 +130,41 @@ func (h HTTPAPI) createEventType(w http.ResponseWriter, r *http.Request, params
 	metricConfigRequests.WithLabelValues(eventType.Space, "eventtype", "create").Inc()
 }
 
+func (h HTTPAPI) updateEventType(w http.ResponseWriter, r *http.Request, params httprouter.Params) {
+	w.Header().Set("Content-Type", "application/json")
+	encoder := json.NewEncoder(w)
+
+	eventType := &event.Type{}
+	dec := json.NewDecoder(r.Body)
+	err := dec.Decode(eventType)
+	if err != nil {
+		w.WriteHeader(http.StatusBadRequest)
+		validationErr := event.ErrEventTypeValidation{Message: err.Error()}
+		encoder.Encode(&Response{Errors: []Error{{Message: validationErr.Error()}}})
+		return
+	}
+
+	eventType.Space = params.ByName("space")
+	eventType.Name = event.TypeName(params.ByName("name"))
+	output, err := h.EventTypes.UpdateEventType(eventType)
+	if err != nil {
+		if _, ok := err.(*event.ErrEventTypeNotFound); ok {
+			w.WriteHeader(http.StatusNotFound)
+		} else if _, ok := err.(*event.ErrEventTypeValidation); ok {
+			w.WriteHeader(http.StatusBadRequest)
+		} else {
+			w.WriteHeader(http.StatusInternalServerError)
+		}
+
+		encoder.Encode(&Response{Errors: []Error{{Message: err.Error()}}})
+	} else {
+		w.WriteHeader(http.StatusOK)
+		encoder.Encode(output)
+	}
+
+	metricConfigRequests.WithLabelValues(eventType.Space, "eventtype", "update").Inc()
+}
+
 func (h HTTPAPI) deleteEventType(w http.ResponseWriter, r *http.Request, params httprouter.Params) {
 	w.Header().Set("Content-Type", "application/json")
 	encoder := json.NewEncoder(w)
@@ -138,7 +174,7 @@ func (h HTTPAPI) deleteEventType(w http.ResponseWriter, r *http.Request, params
 	if err != nil {
 		if _, ok := err.(*event.ErrEventTypeNotFound); ok {
 			w.WriteHeader(http.StatusNotFound)
-		} else if _, ok := err.(*event.ErrEventTypeHasSubscriptionsError); ok {
+		} else if _, ok := err.(*event.ErrEventTypeHasSubscriptions); ok {
 			w.WriteHeader(http.StatusBadRequest)
 		} else {
 			w.WriteHeader(http.StatusInternalServerError)
@@ -270,7 +306,7 @@ func (h HTTPAPI) deleteFunction(w http.ResponseWriter, r *http.Request, params h
 	if err != nil {
 		if _, ok := err.(*function.ErrFunctionNotFound); ok {
 			w.WriteHeader(http.StatusNotFound)
-		} else if _, ok := err.(*function.ErrFunctionHasSubscriptionsError); ok {
+		} else if _, ok := err.(*function.ErrFunctionHasSubscriptions); ok {
 			w.WriteHeader(http.StatusBadRequest)
 		} else {
 			w.WriteHeader(http.StatusInternalServerError)

httpapi/httpapi_test.go

@@ -165,6 +165,73 @@ func TestCreateEventType(t *testing.T) {
 	})
 }
 
+func TestUpdateEventType(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+	router, eventTypes, _, _ := setup(ctrl)
+
+	typePayload := []byte(`{"name":"test.event","space":"test1"}`)
+
+	t.Run("event type updated", func(t *testing.T) {
+		eventType := &event.Type{Space: "default", Name: event.TypeName("test.event")}
+		eventTypes.EXPECT().UpdateEventType(eventType).Return(eventType, nil)
+
+		resp := request(router, http.MethodPut, "/v1/spaces/default/eventtypes/test.event", typePayload)
+
+		returnedType := &event.Type{}
+		json.Unmarshal(resp.Body.Bytes(), returnedType)
+		assert.Equal(t, http.StatusOK, resp.Code)
+		assert.Equal(t, "application/json", resp.Header().Get("Content-Type"))
+		assert.Equal(t, event.TypeName("test.event"), returnedType.Name)
+		assert.Equal(t, "default", returnedType.Space)
+	})
+
+	t.Run("event type doesn't exists", func(t *testing.T) {
+		eventTypes.EXPECT().UpdateEventType(gomock.Any()).
+			Return(nil, &event.ErrEventTypeNotFound{Name: event.TypeName("test.event")})
+
+		resp := request(router, http.MethodPut, "/v1/spaces/default/eventtypes/test.event", typePayload)
+
+		httpresp := &httpapi.Response{}
+		json.Unmarshal(resp.Body.Bytes(), httpresp)
+		assert.Equal(t, http.StatusNotFound, resp.Code)
+		assert.Equal(t, `Event Type "test.event" not found.`, httpresp.Errors[0].Message)
+	})
+
+	t.Run("validation error", func(t *testing.T) {
+		eventTypes.EXPECT().UpdateEventType(gomock.Any()).
+			Return(nil, &event.ErrEventTypeValidation{Message: "some error"})
+
+		payload := []byte(`{"name":"test"}`)
+		resp := request(router, http.MethodPut, "/v1/spaces/default/eventtypes/test.event", payload)
+
+		httpresp := &httpapi.Response{}
+		json.Unmarshal(resp.Body.Bytes(), httpresp)
+		assert.Equal(t, http.StatusBadRequest, resp.Code)
+		assert.Equal(t, "Event Type doesn't validate. Validation error: some error", httpresp.Errors[0].Message)
+	})
+
+	t.Run("malformed JSON", func(t *testing.T) {
+		resp := request(router, http.MethodPut, "/v1/spaces/default/eventtypes/test.event", []byte("{"))
+
+		httpresp := &httpapi.Response{}
+		json.Unmarshal(resp.Body.Bytes(), httpresp)
+		assert.Equal(t, http.StatusBadRequest, resp.Code)
+		assert.Equal(t, "Event Type doesn't validate. Validation error: unexpected EOF", httpresp.Errors[0].Message)
+	})
+
+	t.Run("internal error", func(t *testing.T) {
+		eventTypes.EXPECT().UpdateEventType(gomock.Any()).Return(nil, errors.New("processing error"))
+
+		resp := request(router, http.MethodPut, "/v1/spaces/default/eventtypes/test.event", typePayload)
+
+		httpresp := &httpapi.Response{}
+		json.Unmarshal(resp.Body.Bytes(), httpresp)
+		assert.Equal(t, http.StatusInternalServerError, resp.Code)
+		assert.Equal(t, `processing error`, httpresp.Errors[0].Message)
+	})
+}
+
 func TestDeleteEventType(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	defer ctrl.Finish()
@@ -181,7 +248,7 @@ func TestDeleteEventType(t *testing.T) {
 	})
 
 	t.Run("event type has subscriptions", func(t *testing.T) {
-		eventTypes.EXPECT().DeleteEventType(gomock.Any(), gomock.Any()).Return(&event.ErrEventTypeHasSubscriptionsError{})
+		eventTypes.EXPECT().DeleteEventType(gomock.Any(), gomock.Any()).Return(&event.ErrEventTypeHasSubscriptions{})
 
 		resp := request(router, http.MethodDelete, "/v1/spaces/default/eventtypes/test.event", nil)
 
@@ -388,7 +455,7 @@ func TestDeleteFunction(t *testing.T) {
 	})
 
 	t.Run("function has subscriptions", func(t *testing.T) {
-		functions.EXPECT().DeleteFunction(gomock.Any(), gomock.Any()).Return(&function.ErrFunctionHasSubscriptionsError{})
+		functions.EXPECT().DeleteFunction(gomock.Any(), gomock.Any()).Return(&function.ErrFunctionHasSubscriptions{})
 
 		resp := request(router, http.MethodDelete, "/v1/spaces/default/functions/func1", nil)
 
@@ -520,6 +587,44 @@ func TestUpdateSubscription(t *testing.T) {
 	})
 }
 
+func TestDeleteSubscription(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+	router, _, _, subscriptions := setup(ctrl)
+
+	t.Run("subscription deleted", func(t *testing.T) {
+		subscriptions.EXPECT().DeleteSubscription("default", subscription.ID("testid")).Return(nil)
+
+		resp := request(router, http.MethodDelete, "/v1/spaces/default/subscriptions/testid", nil)
+
+		httpresp := &httpapi.Response{}
+		json.Unmarshal(resp.Body.Bytes(), httpresp)
+		assert.Equal(t, http.StatusNoContent, resp.Code)
+	})
+
+	t.Run("subscriptions not found", func(t *testing.T) {
+		subscriptions.EXPECT().DeleteSubscription(gomock.Any(), gomock.Any()).Return(&subscription.ErrSubscriptionNotFound{ID: subscription.ID("testid")})
+
+		resp := request(router, http.MethodDelete, "/v1/spaces/default/subscriptions/testid", nil)
+
+		httpresp := &httpapi.Response{}
+		json.Unmarshal(resp.Body.Bytes(), httpresp)
+		assert.Equal(t, http.StatusNotFound, resp.Code)
+		assert.Equal(t, `Subscription "testid" not found.`, httpresp.Errors[0].Message)
+	})
+
+	t.Run("internal error", func(t *testing.T) {
+		subscriptions.EXPECT().DeleteSubscription(gomock.Any(), gomock.Any()).Return(errors.New("internal error"))
+
+		resp := request(router, http.MethodDelete, "/v1/spaces/default/subscriptions/testid", nil)
+
+		httpresp := &httpapi.Response{}
+		json.Unmarshal(resp.Body.Bytes(), httpresp)
+		assert.Equal(t, http.StatusInternalServerError, resp.Code)
+		assert.Equal(t, "internal error", httpresp.Errors[0].Message)
+	})
+}
+
 func request(router *httprouter.Router, method string, url string, payload []byte) *httptest.ResponseRecorder {
 	resp := httptest.NewRecorder()
 	body := bytes.NewReader(payload)

libkv/eventtype.go

@@ -24,7 +24,7 @@ func (key EventTypeKey) String() string {
 
 // CreateEventType creates event type in configuration.
 func (service Service) CreateEventType(eventType *event.Type) (*event.Type, error) {
-	if err := service.validateEventType(eventType); err != nil {
+	if err := validateEventType(eventType); err != nil {
 		return nil, err
 	}
 
@@ -97,6 +97,39 @@ func (service Service) GetEventTypes(space string) (event.Types, error) {
 	return event.Types(types), nil
 }
 
+// UpdateEventType updates subscription.
+func (service Service) UpdateEventType(newEventType *event.Type) (*event.Type, error) {
+	if err := validateEventType(newEventType); err != nil {
+		return nil, err
+	}
+
+	_, err := service.GetEventType(newEventType.Space, newEventType.Name)
+	if err != nil {
+		return nil, err
+	}
+
+	if newEventType.AuthorizerID != nil {
+		function, _ := service.GetFunction(newEventType.Space, *newEventType.AuthorizerID)
+		if function == nil {
+			return nil, &event.ErrEventTypeValidation{Message: "Authorizer function doesn't exists."}
+		}
+	}
+
+	buf, err := json.Marshal(newEventType)
+	if err != nil {
+		return nil, &event.ErrEventTypeValidation{Message: err.Error()}
+	}
+
+	err = service.EventTypeStore.Put(EventTypeKey{newEventType.Space, newEventType.Name}.String(), buf, nil)
+	if err != nil {
+		return nil, err
+	}
+
+	service.Log.Debug("Event Type updated.", zap.Object("eventType", newEventType))
+
+	return newEventType, nil
+}
+
 // DeleteEventType deletes event type from the configuration.
 func (service Service) DeleteEventType(space string, name event.TypeName) error {
 	subs, err := service.GetSubscriptions(space)
@@ -105,7 +138,7 @@ func (service Service) DeleteEventType(space string, name event.TypeName) error
 	}
 	for _, sub := range subs {
 		if name == sub.EventType {
-			return &event.ErrEventTypeHasSubscriptionsError{}
+			return &event.ErrEventTypeHasSubscriptions{}
 		}
 	}
 
@@ -119,7 +152,7 @@ func (service Service) DeleteEventType(space string, name event.TypeName) error
 	return nil
 }
 
-func (service Service) validateEventType(eventType *event.Type) error {
+func validateEventType(eventType *event.Type) error {
 	if eventType.Space == "" {
 		eventType.Space = defaultSpace
 	}

libkv/eventtype_test.go

@@ -163,6 +163,70 @@ func TestGetEventTypes(t *testing.T) {
 	})
 }
 
+func TestUpdateEventType(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	existingEventTypeKV := &store.KVPair{Value: []byte(`{"space":"default","name":"test.event"}`)}
+	authorizerID := function.ID("auth")
+	newEventType := &event.Type{Space: "default", Name: "test.event", AuthorizerID: &authorizerID}
+	newEventTypePayload := []byte(`{"space":"default","name":"test.event","authorizerId":"auth"}`)
+	functionKV := &store.KVPair{
+		Value: []byte(`{"functionId":"f1","type":"http","provider":{"url": "http://test.com"}}}`)}
+
+	t.Run("event type updated", func(t *testing.T) {
+		functionsDB := mock.NewMockStore(ctrl)
+		functionsDB.EXPECT().Get("default/auth", gomock.Any()).Return(functionKV, nil)
+		eventTypesDB := mock.NewMockStore(ctrl)
+		eventTypesDB.EXPECT().
+			Get("default/test.event", &store.ReadOptions{Consistent: true}).
+			Return(existingEventTypeKV, nil)
+		eventTypesDB.EXPECT().Put("default/test.event", newEventTypePayload, nil).Return(nil)
+		service := &Service{FunctionStore: functionsDB, EventTypeStore: eventTypesDB, Log: zap.NewNop()}
+
+		_, err := service.UpdateEventType(newEventType)
+
+		assert.Nil(t, err)
+	})
+
+	t.Run("event type not found", func(t *testing.T) {
+		db := mock.NewMockStore(ctrl)
+		db.EXPECT().Get(gomock.Any(), gomock.Any()).Return(nil, errors.New("Key not found in store"))
+		service := &Service{EventTypeStore: db, Log: zap.NewNop()}
+
+		_, err := service.UpdateEventType(newEventType)
+
+		assert.Equal(t, &event.ErrEventTypeNotFound{Name: "test.event"}, err)
+	})
+
+	t.Run("authorizer function doesn't exists error", func(t *testing.T) {
+		functionsDB := mock.NewMockStore(ctrl)
+		functionsDB.EXPECT().Get(gomock.Any(), gomock.Any()).Return(nil, errors.New("Key not found in store"))
+		eventTypesDB := mock.NewMockStore(ctrl)
+		eventTypesDB.EXPECT().Get(gomock.Any(), gomock.Any()).Return(existingEventTypeKV, nil)
+		service := &Service{EventTypeStore: eventTypesDB, FunctionStore: functionsDB, Log: zap.NewNop()}
+
+		_, err := service.UpdateEventType(newEventType)
+
+		assert.Equal(t, &event.ErrEventTypeValidation{
+			Message: "Authorizer function doesn't exists.",
+		}, err)
+	})
+
+	t.Run("KV Put error", func(t *testing.T) {
+		functionsDB := mock.NewMockStore(ctrl)
+		functionsDB.EXPECT().Get(gomock.Any(), gomock.Any()).Return(functionKV, nil)
+		eventTypesDB := mock.NewMockStore(ctrl)
+		eventTypesDB.EXPECT().Get(gomock.Any(), gomock.Any()).Return(existingEventTypeKV, nil)
+		eventTypesDB.EXPECT().Put(gomock.Any(), gomock.Any(), gomock.Any()).Return(errors.New("KV put error"))
+		service := &Service{FunctionStore: functionsDB, EventTypeStore: eventTypesDB, Log: zap.NewNop()}
+
+		_, err := service.UpdateEventType(newEventType)
+
+		assert.EqualError(t, err, "KV put error")
+	})
+}
+
 func TestDeleteEventType(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	defer ctrl.Finish()
@@ -201,6 +265,6 @@ func TestDeleteEventType(t *testing.T) {
 
 		err := service.DeleteEventType("default", event.TypeName("test.event"))
 
-		assert.Equal(t, &event.ErrEventTypeHasSubscriptionsError{}, err)
+		assert.Equal(t, &event.ErrEventTypeHasSubscriptions{}, err)
 	})
 }