From 261e523f45d10d33732e342d8e664d21faa0e5a0 Mon Sep 17 00:00:00 2001
From: Martin Schneppenheim <23424570+weeco@users.noreply.github.com>
Date: Mon, 14 Aug 2023 15:27:07 +0100
Subject: [PATCH] backend: add endpoint to retrieve schema types

---
 backend/pkg/api/handle_schema_registry.go | 22 +++++++++++++++++++++-
 backend/pkg/api/routes.go                 |  1 +
 backend/pkg/console/schema_registry.go    | 14 ++++++++++++++
 backend/pkg/console/servicer.go           |  1 +
 4 files changed, 37 insertions(+), 1 deletion(-)

diff --git a/backend/pkg/api/handle_schema_registry.go b/backend/pkg/api/handle_schema_registry.go
index b592648c7..1edfb1888 100644
--- a/backend/pkg/api/handle_schema_registry.go
+++ b/backend/pkg/api/handle_schema_registry.go
@@ -53,6 +53,26 @@ func (api *API) handleGetSchemaRegistryMode() http.HandlerFunc {
 	}
 }
 
+func (api *API) handleGetSchemaRegistrySchemaTypes() http.HandlerFunc {
+	if !api.Cfg.Kafka.Schema.Enabled {
+		return api.handleSchemaRegistryNotConfigured()
+	}
+
+	return func(w http.ResponseWriter, r *http.Request) {
+		res, err := api.ConsoleSvc.GetSchemaRegistrySchemaTypes(r.Context())
+		if err != nil {
+			rest.SendRESTError(w, r, api.Logger, &rest.Error{
+				Err:      err,
+				Status:   http.StatusBadGateway,
+				Message:  fmt.Sprintf("Failed to retrieve schema registry types from the schema registry: %v", err.Error()),
+				IsSilent: false,
+			})
+			return
+		}
+		rest.SendResponse(w, r, api.Logger, http.StatusOK, res)
+	}
+}
+
 func (api *API) handleGetSchemaRegistryConfig() http.HandlerFunc {
 	if !api.Cfg.Kafka.Schema.Enabled {
 		return api.handleSchemaRegistryNotConfigured()
@@ -64,7 +84,7 @@ func (api *API) handleGetSchemaRegistryConfig() http.HandlerFunc {
 			rest.SendRESTError(w, r, api.Logger, &rest.Error{
 				Err:      err,
 				Status:   http.StatusBadGateway,
-				Message:  fmt.Sprintf("Failed to retrieve schema registry config from the schema registry: %v", err.Error()),
+				Message:  fmt.Sprintf("Failed to retrieve schema registry types from the schema registry: %v", err.Error()),
 				IsSilent: false,
 			})
 			return
diff --git a/backend/pkg/api/routes.go b/backend/pkg/api/routes.go
index ce331d816..300633860 100644
--- a/backend/pkg/api/routes.go
+++ b/backend/pkg/api/routes.go
@@ -136,6 +136,7 @@ func (api *API) routes() *chi.Mux {
 				r.Get("/schema-registry/mode", api.handleGetSchemaRegistryMode())
 				r.Get("/schema-registry/config", api.handleGetSchemaRegistryConfig())
 				r.Get("/schema-registry/subjects", api.handleGetSchemaSubjects())
+				r.Get("/schema-registry/schemas/types", api.handleGetSchemaRegistrySchemaTypes())
 				r.Delete("/schema-registry/subjects/{subject}", api.handleDeleteSubject())
 				r.Delete("/schema-registry/subjects/{subject}/versions/{version}", api.handleDeleteSubjectVersion())
 				r.Get("/schema-registry/subjects/{subject}/versions/{version}", api.handleGetSchemaSubjectDetails())
diff --git a/backend/pkg/console/schema_registry.go b/backend/pkg/console/schema_registry.go
index 5b29aaecf..5e8bd3710 100644
--- a/backend/pkg/console/schema_registry.go
+++ b/backend/pkg/console/schema_registry.go
@@ -332,3 +332,17 @@ func (s *Service) DeleteSchemaRegistrySubjectVersion(ctx context.Context, subjec
 	}
 	return &SchemaRegistryDeleteSubjectVersionResponse{DeletedVersion: res.Version}, nil
 }
+
+// SchemaRegistrySchemaTypes describe the schema types that are supported by the schema registry.
+type SchemaRegistrySchemaTypes struct {
+	SchemaTypes []string `json:"schemaTypes"`
+}
+
+// GetSchemaRegistrySchemaTypes returns the supported schema types.
+func (s *Service) GetSchemaRegistrySchemaTypes(ctx context.Context) (*SchemaRegistrySchemaTypes, error) {
+	res, err := s.kafkaSvc.SchemaService.GetSchemaTypes(ctx)
+	if err != nil {
+		return nil, err
+	}
+	return &SchemaRegistrySchemaTypes{SchemaTypes: res}, nil
+}
diff --git a/backend/pkg/console/servicer.go b/backend/pkg/console/servicer.go
index 1c34c1d41..22b170088 100644
--- a/backend/pkg/console/servicer.go
+++ b/backend/pkg/console/servicer.go
@@ -56,4 +56,5 @@ type Servicer interface {
 	GetSchemaRegistrySubjectDetails(ctx context.Context, subjectName string, version string) (*SchemaRegistrySubjectDetails, error)
 	DeleteSchemaRegistrySubject(ctx context.Context, subjectName string, deletePermanently bool) (*SchemaRegistryDeleteSubjectResponse, error)
 	DeleteSchemaRegistrySubjectVersion(ctx context.Context, subject, version string, deletePermanently bool) (*SchemaRegistryDeleteSubjectVersionResponse, error)
+	GetSchemaRegistrySchemaTypes(ctx context.Context) (*SchemaRegistrySchemaTypes, error)
 }