From 183298e28076848e42e8e8dc8238b75c034ae4c4 Mon Sep 17 00:00:00 2001
From: Xinyi Wang <xinyi.wang@hashicorp.com>
Date: Mon, 17 Jul 2023 13:04:31 -0700
Subject: [PATCH 01/29] init commit

---
 agent/http.go                       |  10 ++
 internal/resource/http/http.go      | 145 ++++++++++++++++++++++++++++
 internal/resource/http/http_test.go |  77 +++++++++++++++
 internal/resource/registry.go       |  13 +++
 4 files changed, 245 insertions(+)
 create mode 100644 internal/resource/http/http.go
 create mode 100644 internal/resource/http/http_test.go

diff --git a/agent/http.go b/agent/http.go
index 32010c343a6c..5c3429a8a73e 100644
--- a/agent/http.go
+++ b/agent/http.go
@@ -36,6 +36,7 @@ import (
 	"github.com/hashicorp/consul/agent/structs"
 	"github.com/hashicorp/consul/agent/uiserver"
 	"github.com/hashicorp/consul/api"
+	resourcehttp "github.com/hashicorp/consul/internal/resource/http"
 	"github.com/hashicorp/consul/lib"
 	"github.com/hashicorp/consul/logging"
 	"github.com/hashicorp/consul/proto/private/pbcommon"
@@ -259,6 +260,15 @@ func (s *HTTPHandlers) handler() http.Handler {
 	handlePProf("/debug/pprof/symbol", pprof.Symbol)
 	handlePProf("/debug/pprof/trace", pprof.Trace)
 
+	mux.Handle("/api/",
+		http.StripPrefix("/api",
+			resourcehttp.NewHandler(
+				s.agent.delegate.ResourceServiceClient(),
+				s.agent.baseDeps.Registry,
+			),
+		),
+	)
+
 	if s.IsUIEnabled() {
 		// Note that we _don't_ support reloading ui_config.{enabled, content_dir,
 		// content_path} since this only runs at initial startup.
diff --git a/internal/resource/http/http.go b/internal/resource/http/http.go
new file mode 100644
index 000000000000..f92b07a71cc1
--- /dev/null
+++ b/internal/resource/http/http.go
@@ -0,0 +1,145 @@
+package http
+
+import (
+	"encoding/json"
+	"fmt"
+	"net/http"
+
+	"google.golang.org/protobuf/encoding/protojson"
+	"google.golang.org/protobuf/types/known/anypb"
+
+	"github.com/hashicorp/consul/internal/resource"
+	"github.com/hashicorp/consul/proto-public/pbresource"
+)
+
+func NewHandler(client pbresource.ResourceServiceClient, registry resource.Registry) http.Handler {
+	mux := http.NewServeMux()
+	for _, t := range registry.Types() {
+		// Individual Resource Endpoints.
+		prefix := fmt.Sprintf("/%s/%s/%s/", t.Type.Group, t.Type.GroupVersion, t.Type.Kind)
+		fmt.Println("REGISTERED URLS: ", prefix)
+		mux.Handle(prefix, http.StripPrefix(prefix, &resourceHandler{t, client}))
+	}
+
+	return mux
+}
+
+type writeRequest struct {
+	// TODO: Owner.
+	Version  string            `json:"version"`
+	Metadata map[string]string `json:"metadata"`
+	Data     json.RawMessage   `json:"data"`
+}
+
+type resourceHandler struct {
+	reg    resource.Registration
+	client pbresource.ResourceServiceClient
+}
+
+func (h *resourceHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	switch r.Method {
+	case http.MethodGet:
+		h.handleRead(w, r)
+	case http.MethodPut:
+		h.handleWrite(w, r)
+	default:
+		w.WriteHeader(http.StatusMethodNotAllowed)
+		return
+	}
+}
+
+func (h *resourceHandler) handleRead(w http.ResponseWriter, r *http.Request) {
+	rsp, err := h.client.Read(r.Context(), &pbresource.ReadRequest{
+		Id: &pbresource.ID{
+			Type:    h.reg.Type,
+			Tenancy: tenancy(r),
+			Name:    r.URL.Path,
+		},
+	})
+
+	if err != nil {
+		w.WriteHeader(http.StatusInternalServerError)
+		return
+	}
+
+	output, err := jsonMarshal(rsp.Resource)
+	if err != nil {
+		w.WriteHeader(http.StatusInternalServerError)
+		return
+	}
+	w.Write(output)
+}
+
+func (h *resourceHandler) handleWrite(w http.ResponseWriter, r *http.Request) {
+	// do we introduce logger in this server?
+	//logger := hclog.New(&hclog.LoggerOptions{Name: "xinyi"})
+	//logger.Debug("DECODING ERROR", "error", err.Error())
+	var req writeRequest
+	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+		w.WriteHeader(http.StatusBadRequest)
+		return
+	}
+
+	data := h.reg.Proto.ProtoReflect().New().Interface()
+	if err := protojson.Unmarshal(req.Data, data); err != nil {
+		fmt.Println("UNMARSHAL REQUEST ERROR", err.Error())
+		w.WriteHeader(http.StatusBadRequest)
+		return
+	}
+
+	a, err := anypb.New(data)
+	if err != nil {
+		w.WriteHeader(http.StatusInternalServerError)
+		return
+	}
+
+	rsp, err := h.client.Write(r.Context(), &pbresource.WriteRequest{
+		Resource: &pbresource.Resource{
+			Id: &pbresource.ID{
+				Type:    h.reg.Type,
+				Tenancy: tenancy(r),
+				Name:    r.URL.Path,
+			},
+			Version:  req.Version,
+			Metadata: req.Metadata,
+			Data:     a,
+		},
+	})
+	if err != nil {
+		fmt.Println("WRITE ERROR", err.Error())
+		w.WriteHeader(http.StatusInternalServerError)
+		return
+	}
+
+	output, err := jsonMarshal(rsp.Resource)
+	if err != nil {
+		fmt.Println("UNMARSHAL RESPONSE ERROR", err.Error())
+		w.WriteHeader(http.StatusInternalServerError)
+		return
+	}
+	w.Write(output)
+}
+
+func tenancy(r *http.Request) *pbresource.Tenancy {
+	// TODO: Read querystring parameters.
+	return &pbresource.Tenancy{
+		Partition: "default",
+		PeerName:  "local",
+		Namespace: "default",
+	}
+}
+
+func jsonMarshal(res *pbresource.Resource) ([]byte, error) {
+	output, err := protojson.Marshal(res)
+	if err != nil {
+		return nil, err
+	}
+
+	var stuff map[string]any
+	if err := json.Unmarshal(output, &stuff); err != nil {
+		return nil, err
+	}
+
+	delete(stuff["data"].(map[string]any), "@type")
+	return json.MarshalIndent(stuff, "", "  ")
+}
diff --git a/internal/resource/http/http_test.go b/internal/resource/http/http_test.go
new file mode 100644
index 000000000000..66d809e018bd
--- /dev/null
+++ b/internal/resource/http/http_test.go
@@ -0,0 +1,77 @@
+package http
+
+import (
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	svctest "github.com/hashicorp/consul/agent/grpc-external/services/resource/testing"
+	"github.com/hashicorp/consul/internal/resource"
+	"github.com/hashicorp/consul/internal/resource/demo"
+	"github.com/hashicorp/consul/proto-public/pbresource"
+	pbdemov2 "github.com/hashicorp/consul/proto/private/pbdemo/v2"
+	"github.com/hashicorp/consul/sdk/testutil"
+)
+
+func TestHandler(t *testing.T) {
+	svc := svctest.RunResourceService(t, demo.RegisterTypes)
+
+	r := resource.NewRegistry()
+	demo.RegisterTypes(r)
+
+	h := NewHandler(svc, r)
+
+	t.Run("Write", func(t *testing.T) {
+		rsp := httptest.NewRecorder()
+		req := httptest.NewRequest("PUT", "/demo/v2/artist/keith-urban", strings.NewReader(`
+			{
+				"metadata": {
+					"foo": "bar"
+				},
+				"data": {
+					"name": "Keith Urban",
+					"genre": "GENRE_COUNTRY"
+				}
+			}
+		`))
+
+		h.ServeHTTP(rsp, req)
+
+		require.Equal(t, http.StatusOK, rsp.Result().StatusCode)
+
+		var result map[string]any
+		require.NoError(t, json.NewDecoder(rsp.Body).Decode(&result))
+		require.Equal(t, "Keith Urban", result["data"].(map[string]any)["name"])
+
+		readRsp, err := svc.Read(testutil.TestContext(t), &pbresource.ReadRequest{
+			Id: &pbresource.ID{
+				Type:    demo.TypeV2Artist,
+				Tenancy: demo.TenancyDefault,
+				Name:    "keith-urban",
+			},
+		})
+		require.NoError(t, err)
+		require.NotNil(t, readRsp.Resource)
+
+		var artist pbdemov2.Artist
+		require.NoError(t, readRsp.Resource.Data.UnmarshalTo(&artist))
+		require.Equal(t, "Keith Urban", artist.Name)
+	})
+
+	t.Run("Read", func(t *testing.T) {
+		rsp := httptest.NewRecorder()
+		req := httptest.NewRequest("GET", "/demo/v2/artist/keith-urban", nil)
+
+		h.ServeHTTP(rsp, req)
+
+		require.Equal(t, http.StatusOK, rsp.Result().StatusCode)
+
+		var result map[string]any
+		require.NoError(t, json.NewDecoder(rsp.Body).Decode(&result))
+		require.Equal(t, "Keith Urban", result["data"].(map[string]any)["name"])
+	})
+}
diff --git a/internal/resource/registry.go b/internal/resource/registry.go
index 0004acfff4c6..4f06c6faa106 100644
--- a/internal/resource/registry.go
+++ b/internal/resource/registry.go
@@ -26,6 +26,8 @@ type Registry interface {
 
 	// Resolve the given resource type and its hooks.
 	Resolve(typ *pbresource.Type) (reg Registration, ok bool)
+
+	Types() []Registration
 }
 
 type Registration struct {
@@ -154,6 +156,17 @@ func (r *TypeRegistry) Resolve(typ *pbresource.Type) (reg Registration, ok bool)
 	return Registration{}, false
 }
 
+func (r *TypeRegistry) Types() []Registration {
+	r.lock.RLock()
+	defer r.lock.RUnlock()
+
+	types := make([]Registration, 0, len(r.registrations))
+	for _, v := range r.registrations {
+		types = append(types, v)
+	}
+	return types
+}
+
 func ToGVK(resourceType *pbresource.Type) string {
 	return fmt.Sprintf("%s.%s.%s", resourceType.Group, resourceType.GroupVersion, resourceType.Kind)
 }

From eaf297d887d649a4aede3b4992554c50899ba140 Mon Sep 17 00:00:00 2001
From: Xinyi Wang <xinyi.wang@hashicorp.com>
Date: Wed, 19 Jul 2023 17:37:55 -0700
Subject: [PATCH 02/29] pass x-consul-token to the grpc server

---
 agent/http.go                  |  1 +
 internal/resource/http/http.go | 26 ++++++++++++++++----------
 2 files changed, 17 insertions(+), 10 deletions(-)

diff --git a/agent/http.go b/agent/http.go
index 5c3429a8a73e..66e346847b44 100644
--- a/agent/http.go
+++ b/agent/http.go
@@ -265,6 +265,7 @@ func (s *HTTPHandlers) handler() http.Handler {
 			resourcehttp.NewHandler(
 				s.agent.delegate.ResourceServiceClient(),
 				s.agent.baseDeps.Registry,
+				s.parseToken,
 			),
 		),
 	)
diff --git a/internal/resource/http/http.go b/internal/resource/http/http.go
index f92b07a71cc1..c0c4a927a20e 100644
--- a/internal/resource/http/http.go
+++ b/internal/resource/http/http.go
@@ -1,10 +1,12 @@
 package http
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
 	"net/http"
 
+	"google.golang.org/grpc/metadata"
 	"google.golang.org/protobuf/encoding/protojson"
 	"google.golang.org/protobuf/types/known/anypb"
 
@@ -12,13 +14,13 @@ import (
 	"github.com/hashicorp/consul/proto-public/pbresource"
 )
 
-func NewHandler(client pbresource.ResourceServiceClient, registry resource.Registry) http.Handler {
+func NewHandler(client pbresource.ResourceServiceClient, registry resource.Registry, parseToken func(req *http.Request, token *string)) http.Handler {
 	mux := http.NewServeMux()
 	for _, t := range registry.Types() {
 		// Individual Resource Endpoints.
 		prefix := fmt.Sprintf("/%s/%s/%s/", t.Type.Group, t.Type.GroupVersion, t.Type.Kind)
 		fmt.Println("REGISTERED URLS: ", prefix)
-		mux.Handle(prefix, http.StripPrefix(prefix, &resourceHandler{t, client}))
+		mux.Handle(prefix, http.StripPrefix(prefix, &resourceHandler{t, client, parseToken}))
 	}
 
 	return mux
@@ -32,24 +34,28 @@ type writeRequest struct {
 }
 
 type resourceHandler struct {
-	reg    resource.Registration
-	client pbresource.ResourceServiceClient
+	reg        resource.Registration
+	client     pbresource.ResourceServiceClient
+	parseToken func(req *http.Request, token *string)
 }
 
 func (h *resourceHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	var token string
+	h.parseToken(r, &token)
+	ctx := metadata.AppendToOutgoingContext(r.Context(), "x-consul-token", token)
 	switch r.Method {
 	case http.MethodGet:
-		h.handleRead(w, r)
+		h.handleRead(w, r, ctx)
 	case http.MethodPut:
-		h.handleWrite(w, r)
+		h.handleWrite(w, r, ctx)
 	default:
 		w.WriteHeader(http.StatusMethodNotAllowed)
 		return
 	}
 }
 
-func (h *resourceHandler) handleRead(w http.ResponseWriter, r *http.Request) {
-	rsp, err := h.client.Read(r.Context(), &pbresource.ReadRequest{
+func (h *resourceHandler) handleRead(w http.ResponseWriter, r *http.Request, ctx context.Context) {
+	rsp, err := h.client.Read(ctx, &pbresource.ReadRequest{
 		Id: &pbresource.ID{
 			Type:    h.reg.Type,
 			Tenancy: tenancy(r),
@@ -70,7 +76,7 @@ func (h *resourceHandler) handleRead(w http.ResponseWriter, r *http.Request) {
 	w.Write(output)
 }
 
-func (h *resourceHandler) handleWrite(w http.ResponseWriter, r *http.Request) {
+func (h *resourceHandler) handleWrite(w http.ResponseWriter, r *http.Request, ctx context.Context) {
 	// do we introduce logger in this server?
 	//logger := hclog.New(&hclog.LoggerOptions{Name: "xinyi"})
 	//logger.Debug("DECODING ERROR", "error", err.Error())
@@ -93,7 +99,7 @@ func (h *resourceHandler) handleWrite(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	rsp, err := h.client.Write(r.Context(), &pbresource.WriteRequest{
+	rsp, err := h.client.Write(ctx, &pbresource.WriteRequest{
 		Resource: &pbresource.Resource{
 			Id: &pbresource.ID{
 				Type:    h.reg.Type,

From 97dacbc29304b0e56873e3f779f394794346a24b Mon Sep 17 00:00:00 2001
From: Xinyi Wang <xinyi.wang@hashicorp.com>
Date: Thu, 20 Jul 2023 10:30:23 -0700
Subject: [PATCH 03/29] query params and add logger

---
 agent/http.go                  |  1 +
 internal/resource/http/http.go | 23 ++++++++++++++++-------
 2 files changed, 17 insertions(+), 7 deletions(-)

diff --git a/agent/http.go b/agent/http.go
index 66e346847b44..31aaf68c2011 100644
--- a/agent/http.go
+++ b/agent/http.go
@@ -266,6 +266,7 @@ func (s *HTTPHandlers) handler() http.Handler {
 				s.agent.delegate.ResourceServiceClient(),
 				s.agent.baseDeps.Registry,
 				s.parseToken,
+				s.agent.logger.Named(logging.HTTP),
 			),
 		),
 	)
diff --git a/internal/resource/http/http.go b/internal/resource/http/http.go
index c0c4a927a20e..548745c8c587 100644
--- a/internal/resource/http/http.go
+++ b/internal/resource/http/http.go
@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"net/http"
 
+	"github.com/hashicorp/go-hclog"
 	"google.golang.org/grpc/metadata"
 	"google.golang.org/protobuf/encoding/protojson"
 	"google.golang.org/protobuf/types/known/anypb"
@@ -14,13 +15,17 @@ import (
 	"github.com/hashicorp/consul/proto-public/pbresource"
 )
 
-func NewHandler(client pbresource.ResourceServiceClient, registry resource.Registry, parseToken func(req *http.Request, token *string)) http.Handler {
+func NewHandler(
+	client pbresource.ResourceServiceClient,
+	registry resource.Registry,
+	parseToken func(req *http.Request, token *string),
+	logger hclog.Logger) http.Handler {
 	mux := http.NewServeMux()
 	for _, t := range registry.Types() {
 		// Individual Resource Endpoints.
 		prefix := fmt.Sprintf("/%s/%s/%s/", t.Type.Group, t.Type.GroupVersion, t.Type.Kind)
 		fmt.Println("REGISTERED URLS: ", prefix)
-		mux.Handle(prefix, http.StripPrefix(prefix, &resourceHandler{t, client, parseToken}))
+		mux.Handle(prefix, http.StripPrefix(prefix, &resourceHandler{t, client, parseToken, logger}))
 	}
 
 	return mux
@@ -37,6 +42,7 @@ type resourceHandler struct {
 	reg        resource.Registration
 	client     pbresource.ResourceServiceClient
 	parseToken func(req *http.Request, token *string)
+	logger     hclog.Logger
 }
 
 func (h *resourceHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
@@ -88,7 +94,6 @@ func (h *resourceHandler) handleWrite(w http.ResponseWriter, r *http.Request, ct
 
 	data := h.reg.Proto.ProtoReflect().New().Interface()
 	if err := protojson.Unmarshal(req.Data, data); err != nil {
-		fmt.Println("UNMARSHAL REQUEST ERROR", err.Error())
 		w.WriteHeader(http.StatusBadRequest)
 		return
 	}
@@ -127,11 +132,15 @@ func (h *resourceHandler) handleWrite(w http.ResponseWriter, r *http.Request, ct
 }
 
 func tenancy(r *http.Request) *pbresource.Tenancy {
-	// TODO: Read querystring parameters.
+	// are partition peername and namespace required fields?
+	params := r.URL.Query()
+	partition := params.Get("partition")
+	peername := params.Get("peer_name")
+	namespace := params.Get("namespace")
 	return &pbresource.Tenancy{
-		Partition: "default",
-		PeerName:  "local",
-		Namespace: "default",
+		Partition: partition,
+		PeerName:  peername,
+		Namespace: namespace,
 	}
 }
 

From 70624293601869c8314bf210fa0ebc09e641a258 Mon Sep 17 00:00:00 2001
From: Xinyi Wang <xinyi.wang@hashicorp.com>
Date: Thu, 20 Jul 2023 11:21:44 -0700
Subject: [PATCH 04/29] log message

---
 internal/resource/http/http.go | 22 +++++++++++-----------
 1 file changed, 11 insertions(+), 11 deletions(-)

diff --git a/internal/resource/http/http.go b/internal/resource/http/http.go
index 548745c8c587..9937af8f2902 100644
--- a/internal/resource/http/http.go
+++ b/internal/resource/http/http.go
@@ -6,11 +6,12 @@ import (
 	"fmt"
 	"net/http"
 
-	"github.com/hashicorp/go-hclog"
 	"google.golang.org/grpc/metadata"
 	"google.golang.org/protobuf/encoding/protojson"
 	"google.golang.org/protobuf/types/known/anypb"
 
+	"github.com/hashicorp/go-hclog"
+
 	"github.com/hashicorp/consul/internal/resource"
 	"github.com/hashicorp/consul/proto-public/pbresource"
 )
@@ -24,7 +25,7 @@ func NewHandler(
 	for _, t := range registry.Types() {
 		// Individual Resource Endpoints.
 		prefix := fmt.Sprintf("/%s/%s/%s/", t.Type.Group, t.Type.GroupVersion, t.Type.Kind)
-		fmt.Println("REGISTERED URLS: ", prefix)
+		logger.Info("Registered resource endpoint: ", prefix)
 		mux.Handle(prefix, http.StripPrefix(prefix, &resourceHandler{t, client, parseToken, logger}))
 	}
 
@@ -83,24 +84,23 @@ func (h *resourceHandler) handleRead(w http.ResponseWriter, r *http.Request, ctx
 }
 
 func (h *resourceHandler) handleWrite(w http.ResponseWriter, r *http.Request, ctx context.Context) {
-	// do we introduce logger in this server?
-	//logger := hclog.New(&hclog.LoggerOptions{Name: "xinyi"})
-	//logger.Debug("DECODING ERROR", "error", err.Error())
 	var req writeRequest
+	// convert req data to struct
 	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
 		w.WriteHeader(http.StatusBadRequest)
 		return
 	}
-
+	// struct to proto message
 	data := h.reg.Proto.ProtoReflect().New().Interface()
 	if err := protojson.Unmarshal(req.Data, data); err != nil {
 		w.WriteHeader(http.StatusBadRequest)
 		return
 	}
-
-	a, err := anypb.New(data)
+	// proto message to any
+	anyProtoMsg, err := anypb.New(data)
 	if err != nil {
 		w.WriteHeader(http.StatusInternalServerError)
+		h.logger.Error("Failed to convert proto message to any type: ", err)
 		return
 	}
 
@@ -113,19 +113,19 @@ func (h *resourceHandler) handleWrite(w http.ResponseWriter, r *http.Request, ct
 			},
 			Version:  req.Version,
 			Metadata: req.Metadata,
-			Data:     a,
+			Data:     anyProtoMsg,
 		},
 	})
 	if err != nil {
-		fmt.Println("WRITE ERROR", err.Error())
 		w.WriteHeader(http.StatusInternalServerError)
+		h.logger.Error("Failed to write to GRPC resource: ", err)
 		return
 	}
 
 	output, err := jsonMarshal(rsp.Resource)
 	if err != nil {
-		fmt.Println("UNMARSHAL RESPONSE ERROR", err.Error())
 		w.WriteHeader(http.StatusInternalServerError)
+		h.logger.Error("Failed to unmarshal GRPC resource response: ", err)
 		return
 	}
 	w.Write(output)

From e69aa2dd64bb1d716e82a792132b63bf7e6f17dc Mon Sep 17 00:00:00 2001
From: Xinyi Wang <xinyi.wang@hashicorp.com>
Date: Thu, 20 Jul 2023 11:29:34 -0700
Subject: [PATCH 05/29] change log message

---
 internal/resource/http/http.go | 8 ++++----
 1 file changed, 4 insertions(+), 4 deletions(-)

diff --git a/internal/resource/http/http.go b/internal/resource/http/http.go
index 9937af8f2902..d40b719d8112 100644
--- a/internal/resource/http/http.go
+++ b/internal/resource/http/http.go
@@ -25,7 +25,7 @@ func NewHandler(
 	for _, t := range registry.Types() {
 		// Individual Resource Endpoints.
 		prefix := fmt.Sprintf("/%s/%s/%s/", t.Type.Group, t.Type.GroupVersion, t.Type.Kind)
-		logger.Info("Registered resource endpoint: ", prefix)
+		logger.Info("Registered resource endpoint", "endpoint", prefix)
 		mux.Handle(prefix, http.StripPrefix(prefix, &resourceHandler{t, client, parseToken, logger}))
 	}
 
@@ -100,7 +100,7 @@ func (h *resourceHandler) handleWrite(w http.ResponseWriter, r *http.Request, ct
 	anyProtoMsg, err := anypb.New(data)
 	if err != nil {
 		w.WriteHeader(http.StatusInternalServerError)
-		h.logger.Error("Failed to convert proto message to any type: ", err)
+		h.logger.Error("Failed to convert proto message to any type", "error", err)
 		return
 	}
 
@@ -118,14 +118,14 @@ func (h *resourceHandler) handleWrite(w http.ResponseWriter, r *http.Request, ct
 	})
 	if err != nil {
 		w.WriteHeader(http.StatusInternalServerError)
-		h.logger.Error("Failed to write to GRPC resource: ", err)
+		h.logger.Error("Failed to write to GRPC resource", "error", err)
 		return
 	}
 
 	output, err := jsonMarshal(rsp.Resource)
 	if err != nil {
 		w.WriteHeader(http.StatusInternalServerError)
-		h.logger.Error("Failed to unmarshal GRPC resource response: ", err)
+		h.logger.Error("Failed to unmarshal GRPC resource response", "error", err)
 		return
 	}
 	w.Write(output)

From 4975291f45fbcb69277f1c58461c9d1d106b7568 Mon Sep 17 00:00:00 2001
From: Xinyi Wang <xinyi.wang@hashicorp.com>
Date: Thu, 20 Jul 2023 15:04:54 -0700
Subject: [PATCH 06/29] fix unit test

---
 internal/resource/http/http.go      | 50 +++++++++++++++++++++--------
 internal/resource/http/http_test.go | 37 +++++++++------------
 2 files changed, 51 insertions(+), 36 deletions(-)

diff --git a/internal/resource/http/http.go b/internal/resource/http/http.go
index d40b719d8112..161a859f73b3 100644
--- a/internal/resource/http/http.go
+++ b/internal/resource/http/http.go
@@ -5,6 +5,8 @@ import (
 	"encoding/json"
 	"fmt"
 	"net/http"
+	"path"
+	"strings"
 
 	"google.golang.org/grpc/metadata"
 	"google.golang.org/protobuf/encoding/protojson"
@@ -24,7 +26,7 @@ func NewHandler(
 	mux := http.NewServeMux()
 	for _, t := range registry.Types() {
 		// Individual Resource Endpoints.
-		prefix := fmt.Sprintf("/%s/%s/%s/", t.Type.Group, t.Type.GroupVersion, t.Type.Kind)
+		prefix := strings.ToLower(fmt.Sprintf("/%s/%s/%s/", t.Type.Group, t.Type.GroupVersion, t.Type.Kind))
 		logger.Info("Registered resource endpoint", "endpoint", prefix)
 		mux.Handle(prefix, http.StripPrefix(prefix, &resourceHandler{t, client, parseToken, logger}))
 	}
@@ -62,10 +64,15 @@ func (h *resourceHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 }
 
 func (h *resourceHandler) handleRead(w http.ResponseWriter, r *http.Request, ctx context.Context) {
+	tenancyInfo, _ := checkURL(r)
+	if tenancyInfo == nil {
+		w.WriteHeader(http.StatusBadRequest)
+		return
+	}
 	rsp, err := h.client.Read(ctx, &pbresource.ReadRequest{
 		Id: &pbresource.ID{
 			Type:    h.reg.Type,
-			Tenancy: tenancy(r),
+			Tenancy: tenancyInfo,
 			Name:    r.URL.Path,
 		},
 	})
@@ -88,13 +95,13 @@ func (h *resourceHandler) handleWrite(w http.ResponseWriter, r *http.Request, ct
 	// convert req data to struct
 	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
 		w.WriteHeader(http.StatusBadRequest)
-		return
+		w.Write([]byte("Request body didn't follow schema."))
 	}
 	// struct to proto message
 	data := h.reg.Proto.ProtoReflect().New().Interface()
 	if err := protojson.Unmarshal(req.Data, data); err != nil {
-		w.WriteHeader(http.StatusBadRequest)
-		return
+		w.WriteHeader(http.StatusInternalServerError)
+		w.Write([]byte("Request body didn't follow schema."))
 	}
 	// proto message to any
 	anyProtoMsg, err := anypb.New(data)
@@ -104,12 +111,21 @@ func (h *resourceHandler) handleWrite(w http.ResponseWriter, r *http.Request, ct
 		return
 	}
 
+	tenancyInfo, resourceName := checkURL(r)
+	if tenancyInfo == nil {
+		w.WriteHeader(http.StatusBadRequest)
+		w.Write([]byte("Missing partition, peer_name or namespace in the query params"))
+	}
+	if resourceName == "" {
+		w.WriteHeader(http.StatusBadRequest)
+		w.Write([]byte("Missing resource name in the URL"))
+	}
 	rsp, err := h.client.Write(ctx, &pbresource.WriteRequest{
 		Resource: &pbresource.Resource{
 			Id: &pbresource.ID{
 				Type:    h.reg.Type,
-				Tenancy: tenancy(r),
-				Name:    r.URL.Path,
+				Tenancy: tenancyInfo,
+				Name:    resourceName,
 			},
 			Version:  req.Version,
 			Metadata: req.Metadata,
@@ -131,17 +147,23 @@ func (h *resourceHandler) handleWrite(w http.ResponseWriter, r *http.Request, ct
 	w.Write(output)
 }
 
-func tenancy(r *http.Request) *pbresource.Tenancy {
-	// are partition peername and namespace required fields?
+func checkURL(r *http.Request) (tenancy *pbresource.Tenancy, resourceName string) {
 	params := r.URL.Query()
 	partition := params.Get("partition")
-	peername := params.Get("peer_name")
+	peerName := params.Get("peer_name")
 	namespace := params.Get("namespace")
-	return &pbresource.Tenancy{
-		Partition: partition,
-		PeerName:  peername,
-		Namespace: namespace,
+	if partition == "" || peerName == "" || namespace == "" {
+		tenancy = nil
+	} else {
+		tenancy = &pbresource.Tenancy{
+			Partition: partition,
+			PeerName:  peerName,
+			Namespace: namespace,
+		}
 	}
+	resourceName = path.Base(r.URL.Path)
+
+	return
 }
 
 func jsonMarshal(res *pbresource.Resource) ([]byte, error) {
diff --git a/internal/resource/http/http_test.go b/internal/resource/http/http_test.go
index 66d809e018bd..6bc9be9a2cf0 100644
--- a/internal/resource/http/http_test.go
+++ b/internal/resource/http/http_test.go
@@ -7,6 +7,7 @@ import (
 	"strings"
 	"testing"
 
+	"github.com/hashicorp/go-hclog"
 	"github.com/stretchr/testify/require"
 
 	svctest "github.com/hashicorp/consul/agent/grpc-external/services/resource/testing"
@@ -17,17 +18,22 @@ import (
 	"github.com/hashicorp/consul/sdk/testutil"
 )
 
-func TestHandler(t *testing.T) {
-	svc := svctest.RunResourceService(t, demo.RegisterTypes)
+func TestResourceHandler(t *testing.T) {
+	client := svctest.RunResourceService(t, demo.RegisterTypes)
 
-	r := resource.NewRegistry()
-	demo.RegisterTypes(r)
-
-	h := NewHandler(svc, r)
+	resourceHandler := resourceHandler{
+		resource.Registration{
+			Type:  demo.TypeV2Artist,
+			Proto: &pbdemov2.Artist{},
+		},
+		client,
+		func(req *http.Request, token *string) { return },
+		hclog.NewNullLogger(),
+	}
 
 	t.Run("Write", func(t *testing.T) {
 		rsp := httptest.NewRecorder()
-		req := httptest.NewRequest("PUT", "/demo/v2/artist/keith-urban", strings.NewReader(`
+		req := httptest.NewRequest("PUT", "/api/demo/v2/artist/keith-urban?partition=default&peer_name=local&namespace=default", strings.NewReader(`
 			{
 				"metadata": {
 					"foo": "bar"
@@ -39,7 +45,7 @@ func TestHandler(t *testing.T) {
 			}
 		`))
 
-		h.ServeHTTP(rsp, req)
+		resourceHandler.ServeHTTP(rsp, req)
 
 		require.Equal(t, http.StatusOK, rsp.Result().StatusCode)
 
@@ -47,7 +53,7 @@ func TestHandler(t *testing.T) {
 		require.NoError(t, json.NewDecoder(rsp.Body).Decode(&result))
 		require.Equal(t, "Keith Urban", result["data"].(map[string]any)["name"])
 
-		readRsp, err := svc.Read(testutil.TestContext(t), &pbresource.ReadRequest{
+		readRsp, err := client.Read(testutil.TestContext(t), &pbresource.ReadRequest{
 			Id: &pbresource.ID{
 				Type:    demo.TypeV2Artist,
 				Tenancy: demo.TenancyDefault,
@@ -61,17 +67,4 @@ func TestHandler(t *testing.T) {
 		require.NoError(t, readRsp.Resource.Data.UnmarshalTo(&artist))
 		require.Equal(t, "Keith Urban", artist.Name)
 	})
-
-	t.Run("Read", func(t *testing.T) {
-		rsp := httptest.NewRecorder()
-		req := httptest.NewRequest("GET", "/demo/v2/artist/keith-urban", nil)
-
-		h.ServeHTTP(rsp, req)
-
-		require.Equal(t, http.StatusOK, rsp.Result().StatusCode)
-
-		var result map[string]any
-		require.NoError(t, json.NewDecoder(rsp.Body).Decode(&result))
-		require.Equal(t, "Keith Urban", result["data"].(map[string]any)["name"])
-	})
 }

From f5e9707a849d0c883a848889d88790f4accee560 Mon Sep 17 00:00:00 2001
From: Xinyi Wang <xinyi.wang@hashicorp.com>
Date: Fri, 21 Jul 2023 08:29:18 -0700
Subject: [PATCH 07/29] remove read

---
 internal/resource/http/http.go | 29 -----------------------------
 1 file changed, 29 deletions(-)

diff --git a/internal/resource/http/http.go b/internal/resource/http/http.go
index 161a859f73b3..870dc5c6eaab 100644
--- a/internal/resource/http/http.go
+++ b/internal/resource/http/http.go
@@ -53,8 +53,6 @@ func (h *resourceHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	h.parseToken(r, &token)
 	ctx := metadata.AppendToOutgoingContext(r.Context(), "x-consul-token", token)
 	switch r.Method {
-	case http.MethodGet:
-		h.handleRead(w, r, ctx)
 	case http.MethodPut:
 		h.handleWrite(w, r, ctx)
 	default:
@@ -63,33 +61,6 @@ func (h *resourceHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
-func (h *resourceHandler) handleRead(w http.ResponseWriter, r *http.Request, ctx context.Context) {
-	tenancyInfo, _ := checkURL(r)
-	if tenancyInfo == nil {
-		w.WriteHeader(http.StatusBadRequest)
-		return
-	}
-	rsp, err := h.client.Read(ctx, &pbresource.ReadRequest{
-		Id: &pbresource.ID{
-			Type:    h.reg.Type,
-			Tenancy: tenancyInfo,
-			Name:    r.URL.Path,
-		},
-	})
-
-	if err != nil {
-		w.WriteHeader(http.StatusInternalServerError)
-		return
-	}
-
-	output, err := jsonMarshal(rsp.Resource)
-	if err != nil {
-		w.WriteHeader(http.StatusInternalServerError)
-		return
-	}
-	w.Write(output)
-}
-
 func (h *resourceHandler) handleWrite(w http.ResponseWriter, r *http.Request, ctx context.Context) {
 	var req writeRequest
 	// convert req data to struct

From f1bfb5d788caa5797b629da54f5c339c5401829d Mon Sep 17 00:00:00 2001
From: Xinyi Wang <xinyi.wang@hashicorp.com>
Date: Fri, 21 Jul 2023 10:22:13 -0700
Subject: [PATCH 08/29] add detailed error handling message

---
 internal/resource/http/http.go | 23 +++++++++++++++++++++--
 1 file changed, 21 insertions(+), 2 deletions(-)

diff --git a/internal/resource/http/http.go b/internal/resource/http/http.go
index 870dc5c6eaab..92e8ad0d9258 100644
--- a/internal/resource/http/http.go
+++ b/internal/resource/http/http.go
@@ -8,7 +8,9 @@ import (
 	"path"
 	"strings"
 
+	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/metadata"
+	"google.golang.org/grpc/status"
 	"google.golang.org/protobuf/encoding/protojson"
 	"google.golang.org/protobuf/types/known/anypb"
 
@@ -104,8 +106,25 @@ func (h *resourceHandler) handleWrite(w http.ResponseWriter, r *http.Request, ct
 		},
 	})
 	if err != nil {
-		w.WriteHeader(http.StatusInternalServerError)
-		h.logger.Error("Failed to write to GRPC resource", "error", err)
+		if e, ok := status.FromError(err); ok {
+			switch e.Code() {
+			case codes.PermissionDenied:
+				w.WriteHeader(http.StatusForbidden)
+				h.logger.Info("Failed to write to GRPC resource: User not authenticated", "error", err)
+			case codes.Internal:
+				w.WriteHeader(http.StatusInternalServerError)
+				h.logger.Error("Failed to write to GRPC resource: Internal error", "error", err)
+			case codes.Aborted:
+				w.WriteHeader(http.StatusInternalServerError)
+				h.logger.Error("Failed to write to GRPC resource: GRPC aborted", "error", err)
+			default:
+				w.WriteHeader(http.StatusInternalServerError)
+				h.logger.Error("Failed to write to GRPC resource", "error", err)
+			}
+		} else {
+			w.WriteHeader(http.StatusInternalServerError)
+			h.logger.Error("Failed to write to GRPC resource: not able to parse error returned", "error", err)
+		}
 		return
 	}
 

From 42d6ccfa513a80f83fa5847e96c21a6e5bfe77d0 Mon Sep 17 00:00:00 2001
From: Xinyi Wang <xinyi.wang@hashicorp.com>
Date: Fri, 21 Jul 2023 12:35:43 -0700
Subject: [PATCH 09/29] fix unit tests

---
 agent/agent_endpoint_test.go | 29 +++++++++++++++++++++++++----
 1 file changed, 25 insertions(+), 4 deletions(-)

diff --git a/agent/agent_endpoint_test.go b/agent/agent_endpoint_test.go
index 1a275f61afb3..152a9b954015 100644
--- a/agent/agent_endpoint_test.go
+++ b/agent/agent_endpoint_test.go
@@ -1601,14 +1601,35 @@ func TestAgent_Metrics_ACLDeny(t *testing.T) {
 	})
 }
 
+func newDefaultBaseDeps(t *testing.T) BaseDeps {
+	dataDir := testutil.TempDir(t, "acl-agent")
+	logBuffer := testutil.NewLogBuffer(t)
+	logger := hclog.NewInterceptLogger(nil)
+	loader := func(source config.Source) (config.LoadResult, error) {
+		dataDir := fmt.Sprintf(`data_dir = "%s"`, dataDir)
+		opts := config.LoadOpts{
+			HCL:           []string{TestConfigHCL(NodeID()), "", dataDir},
+			DefaultConfig: source,
+		}
+		result, err := config.Load(opts)
+		if result.RuntimeConfig != nil {
+			result.RuntimeConfig.Telemetry.Disable = true
+		}
+		return result, err
+	}
+	bd, err := NewBaseDeps(loader, logBuffer, logger)
+	require.NoError(t, err)
+	return bd
+}
+
 func TestHTTPHandlers_AgentMetricsStream_ACLDeny(t *testing.T) {
-	bd := BaseDeps{}
+	bd := newDefaultBaseDeps(t)
 	bd.Tokens = new(tokenStore.Store)
 	sink := metrics.NewInmemSink(30*time.Millisecond, time.Second)
 	bd.MetricsConfig = &lib.MetricsConfig{
 		Handler: sink,
 	}
-	d := fakeResolveTokenDelegate{authorizer: acl.DenyAll()}
+	d := fakeResolveTokenDelegate{delegate: &delegateMock{}, authorizer: acl.DenyAll()}
 	agent := &Agent{
 		baseDeps: bd,
 		delegate: d,
@@ -1631,13 +1652,13 @@ func TestHTTPHandlers_AgentMetricsStream_ACLDeny(t *testing.T) {
 }
 
 func TestHTTPHandlers_AgentMetricsStream(t *testing.T) {
-	bd := BaseDeps{}
+	bd := newDefaultBaseDeps(t)
 	bd.Tokens = new(tokenStore.Store)
 	sink := metrics.NewInmemSink(20*time.Millisecond, time.Second)
 	bd.MetricsConfig = &lib.MetricsConfig{
 		Handler: sink,
 	}
-	d := fakeResolveTokenDelegate{authorizer: acl.ManageAll()}
+	d := fakeResolveTokenDelegate{delegate: &delegateMock{}, authorizer: acl.ManageAll()}
 	agent := &Agent{
 		baseDeps: bd,
 		delegate: d,

From 595ac3ed8a9b8f90b27dddb4382b2698645a7d69 Mon Sep 17 00:00:00 2001
From: Xinyi Wang <xinyi.wang@hashicorp.com>
Date: Fri, 21 Jul 2023 14:05:27 -0700
Subject: [PATCH 10/29] fix unit tests

---
 agent/agent_test.go | 11 +++++++++++
 1 file changed, 11 insertions(+)

diff --git a/agent/agent_test.go b/agent/agent_test.go
index a2e27feaf4fd..646cc340595a 100644
--- a/agent/agent_test.go
+++ b/agent/agent_test.go
@@ -57,6 +57,7 @@ import (
 	"github.com/hashicorp/consul/agent/token"
 	"github.com/hashicorp/consul/api"
 	"github.com/hashicorp/consul/internal/go-sso/oidcauth/oidcauthtest"
+	"github.com/hashicorp/consul/internal/resource"
 	"github.com/hashicorp/consul/ipaddr"
 	"github.com/hashicorp/consul/lib"
 	"github.com/hashicorp/consul/proto/private/pbautoconf"
@@ -322,6 +323,7 @@ func TestAgent_HTTPMaxHeaderBytes(t *testing.T) {
 					Tokens:          new(token.Store),
 					TLSConfigurator: tlsConf,
 					GRPCConnPool:    &fakeGRPCConnPool{},
+					Registry:        resource.NewRegistry(),
 				},
 				RuntimeConfig: &config.RuntimeConfig{
 					HTTPAddrs: []net.Addr{
@@ -344,6 +346,7 @@ func TestAgent_HTTPMaxHeaderBytes(t *testing.T) {
 			require.NoError(t, err)
 
 			a, err := New(bd)
+			a.delegate = &delegateMock{}
 			require.NoError(t, err)
 
 			a.startLicenseManager(testutil.TestContext(t))
@@ -5477,6 +5480,7 @@ func TestAgent_ListenHTTP_MultipleAddresses(t *testing.T) {
 			Tokens:          new(token.Store),
 			TLSConfigurator: tlsConf,
 			GRPCConnPool:    &fakeGRPCConnPool{},
+			Registry:        resource.NewRegistry(),
 		},
 		RuntimeConfig: &config.RuntimeConfig{
 			HTTPAddrs: []net.Addr{
@@ -5499,6 +5503,7 @@ func TestAgent_ListenHTTP_MultipleAddresses(t *testing.T) {
 	require.NoError(t, err)
 
 	agent, err := New(bd)
+	agent.delegate = &delegateMock{}
 	require.NoError(t, err)
 
 	agent.startLicenseManager(testutil.TestContext(t))
@@ -6073,6 +6078,7 @@ func TestAgent_startListeners(t *testing.T) {
 			Logger:       hclog.NewInterceptLogger(nil),
 			Tokens:       new(token.Store),
 			GRPCConnPool: &fakeGRPCConnPool{},
+			Registry:     resource.NewRegistry(),
 		},
 		RuntimeConfig: &config.RuntimeConfig{
 			HTTPAddrs: []net.Addr{},
@@ -6091,6 +6097,7 @@ func TestAgent_startListeners(t *testing.T) {
 	require.NoError(t, err)
 
 	agent, err := New(bd)
+	agent.delegate = &delegateMock{}
 	require.NoError(t, err)
 
 	// use up an address
@@ -6213,6 +6220,7 @@ func TestAgent_startListeners_scada(t *testing.T) {
 			HCP: hcp.Deps{
 				Provider: pvd,
 			},
+			Registry: resource.NewRegistry(),
 		},
 		RuntimeConfig: &config.RuntimeConfig{},
 		Cache:         cache.New(cache.Options{}),
@@ -6230,6 +6238,7 @@ func TestAgent_startListeners_scada(t *testing.T) {
 	require.NoError(t, err)
 
 	agent, err := New(bd)
+	agent.delegate = &delegateMock{}
 	require.NoError(t, err)
 
 	_, err = agent.startListeners([]net.Addr{c})
@@ -6273,6 +6282,7 @@ func TestAgent_checkServerLastSeen(t *testing.T) {
 			Logger:       hclog.NewInterceptLogger(nil),
 			Tokens:       new(token.Store),
 			GRPCConnPool: &fakeGRPCConnPool{},
+			Registry:     resource.NewRegistry(),
 		},
 		RuntimeConfig: &config.RuntimeConfig{},
 		Cache:         cache.New(cache.Options{}),
@@ -6284,6 +6294,7 @@ func TestAgent_checkServerLastSeen(t *testing.T) {
 		Config:      leafcert.Config{},
 	})
 	agent, err := New(bd)
+	agent.delegate = &delegateMock{}
 	require.NoError(t, err)
 
 	// Test that an ErrNotExist OS error is treated as ok.

From 4ccddf3dc5a9099dec7dd06ac715311dbe54eadd Mon Sep 17 00:00:00 2001
From: Xinyi Wang <xinyi.wang@hashicorp.com>
Date: Mon, 24 Jul 2023 15:54:47 -0700
Subject: [PATCH 11/29] read endpoint

---
 internal/resource/http/http.go      | 78 ++++++++++++++++++++++-------
 internal/resource/http/http_test.go | 15 +++++-
 2 files changed, 73 insertions(+), 20 deletions(-)

diff --git a/internal/resource/http/http.go b/internal/resource/http/http.go
index 92e8ad0d9258..9909602ade81 100644
--- a/internal/resource/http/http.go
+++ b/internal/resource/http/http.go
@@ -57,6 +57,8 @@ func (h *resourceHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	switch r.Method {
 	case http.MethodPut:
 		h.handleWrite(w, r, ctx)
+	case http.MethodGet:
+		h.handleRead(w, r, ctx)
 	default:
 		w.WriteHeader(http.StatusMethodNotAllowed)
 		return
@@ -106,25 +108,38 @@ func (h *resourceHandler) handleWrite(w http.ResponseWriter, r *http.Request, ct
 		},
 	})
 	if err != nil {
-		if e, ok := status.FromError(err); ok {
-			switch e.Code() {
-			case codes.PermissionDenied:
-				w.WriteHeader(http.StatusForbidden)
-				h.logger.Info("Failed to write to GRPC resource: User not authenticated", "error", err)
-			case codes.Internal:
-				w.WriteHeader(http.StatusInternalServerError)
-				h.logger.Error("Failed to write to GRPC resource: Internal error", "error", err)
-			case codes.Aborted:
-				w.WriteHeader(http.StatusInternalServerError)
-				h.logger.Error("Failed to write to GRPC resource: GRPC aborted", "error", err)
-			default:
-				w.WriteHeader(http.StatusInternalServerError)
-				h.logger.Error("Failed to write to GRPC resource", "error", err)
-			}
-		} else {
-			w.WriteHeader(http.StatusInternalServerError)
-			h.logger.Error("Failed to write to GRPC resource: not able to parse error returned", "error", err)
-		}
+		handleResponseError(err, w, h)
+		return
+	}
+
+	output, err := jsonMarshal(rsp.Resource)
+	if err != nil {
+		w.WriteHeader(http.StatusInternalServerError)
+		h.logger.Error("Failed to unmarshal GRPC resource response", "error", err)
+		return
+	}
+	w.Write(output)
+}
+
+func (h *resourceHandler) handleRead(w http.ResponseWriter, r *http.Request, ctx context.Context) {
+	tenancyInfo, resourceName := checkURL(r)
+	if tenancyInfo == nil {
+		w.WriteHeader(http.StatusBadRequest)
+		w.Write([]byte("Missing partition, peer_name or namespace in the query params"))
+	}
+	if resourceName == "" {
+		w.WriteHeader(http.StatusBadRequest)
+		w.Write([]byte("Missing resource name in the URL"))
+	}
+	rsp, err := h.client.Read(ctx, &pbresource.ReadRequest{
+		Id: &pbresource.ID{
+			Type:    h.reg.Type,
+			Tenancy: tenancyInfo,
+			Name:    resourceName,
+		},
+	})
+	if err != nil {
+		handleResponseError(err, w, h)
 		return
 	}
 
@@ -170,3 +185,28 @@ func jsonMarshal(res *pbresource.Resource) ([]byte, error) {
 	delete(stuff["data"].(map[string]any), "@type")
 	return json.MarshalIndent(stuff, "", "  ")
 }
+
+func handleResponseError(err error, w http.ResponseWriter, h *resourceHandler) {
+	if e, ok := status.FromError(err); ok {
+		switch e.Code() {
+		case codes.PermissionDenied:
+			w.WriteHeader(http.StatusForbidden)
+			h.logger.Info("Failed to write to GRPC resource: User not authenticated", "error", err)
+		case codes.Internal:
+			w.WriteHeader(http.StatusInternalServerError)
+			h.logger.Error("Failed to write to GRPC resource: Internal error", "error", err)
+		case codes.Aborted:
+			w.WriteHeader(http.StatusInternalServerError)
+			h.logger.Error("Failed to write to GRPC resource: GRPC aborted", "error", err)
+		case codes.NotFound:
+			w.WriteHeader(http.StatusNotFound)
+			h.logger.Info("Failed to write to GRPC resource: Not found", "error", err)
+		default:
+			w.WriteHeader(http.StatusInternalServerError)
+			h.logger.Error("Failed to write to GRPC resource", "error", err)
+		}
+	} else {
+		w.WriteHeader(http.StatusInternalServerError)
+		h.logger.Error("Failed to write to GRPC resource: not able to parse error returned", "error", err)
+	}
+}
diff --git a/internal/resource/http/http_test.go b/internal/resource/http/http_test.go
index 6bc9be9a2cf0..9ece681e4b01 100644
--- a/internal/resource/http/http_test.go
+++ b/internal/resource/http/http_test.go
@@ -33,7 +33,7 @@ func TestResourceHandler(t *testing.T) {
 
 	t.Run("Write", func(t *testing.T) {
 		rsp := httptest.NewRecorder()
-		req := httptest.NewRequest("PUT", "/api/demo/v2/artist/keith-urban?partition=default&peer_name=local&namespace=default", strings.NewReader(`
+		req := httptest.NewRequest("PUT", "/demo/v2/artist/keith-urban?partition=default&peer_name=local&namespace=default", strings.NewReader(`
 			{
 				"metadata": {
 					"foo": "bar"
@@ -67,4 +67,17 @@ func TestResourceHandler(t *testing.T) {
 		require.NoError(t, readRsp.Resource.Data.UnmarshalTo(&artist))
 		require.Equal(t, "Keith Urban", artist.Name)
 	})
+
+	t.Run("Read", func(t *testing.T) {
+		rsp := httptest.NewRecorder()
+		req := httptest.NewRequest("GET", "/demo/v2/artist/keith-urban?partition=default&peer_name=local&namespace=default", nil)
+
+		resourceHandler.ServeHTTP(rsp, req)
+
+		require.Equal(t, http.StatusOK, rsp.Result().StatusCode)
+
+		var result map[string]any
+		require.NoError(t, json.NewDecoder(rsp.Body).Decode(&result))
+		require.Equal(t, "Keith Urban", result["data"].(map[string]any)["name"])
+	})
 }

From eb96780e9b071e108e16c4434c3eff6a4cb7667a Mon Sep 17 00:00:00 2001
From: Xinyi Wang <xinyi.wang@hashicorp.com>
Date: Tue, 25 Jul 2023 11:26:47 -0700
Subject: [PATCH 12/29] general refactor

---
 internal/resource/http/http.go | 43 +++++++++++++++++-----------------
 1 file changed, 22 insertions(+), 21 deletions(-)

diff --git a/internal/resource/http/http.go b/internal/resource/http/http.go
index 92e8ad0d9258..7a988eb7961b 100644
--- a/internal/resource/http/http.go
+++ b/internal/resource/http/http.go
@@ -73,7 +73,7 @@ func (h *resourceHandler) handleWrite(w http.ResponseWriter, r *http.Request, ct
 	// struct to proto message
 	data := h.reg.Proto.ProtoReflect().New().Interface()
 	if err := protojson.Unmarshal(req.Data, data); err != nil {
-		w.WriteHeader(http.StatusInternalServerError)
+		w.WriteHeader(http.StatusBadRequest)
 		w.Write([]byte("Request body didn't follow schema."))
 	}
 	// proto message to any
@@ -87,7 +87,7 @@ func (h *resourceHandler) handleWrite(w http.ResponseWriter, r *http.Request, ct
 	tenancyInfo, resourceName := checkURL(r)
 	if tenancyInfo == nil {
 		w.WriteHeader(http.StatusBadRequest)
-		w.Write([]byte("Missing partition, peer_name or namespace in the query params"))
+		w.Write([]byte("Query params partition, peer_name, and namespace are required."))
 	}
 	if resourceName == "" {
 		w.WriteHeader(http.StatusBadRequest)
@@ -106,25 +106,7 @@ func (h *resourceHandler) handleWrite(w http.ResponseWriter, r *http.Request, ct
 		},
 	})
 	if err != nil {
-		if e, ok := status.FromError(err); ok {
-			switch e.Code() {
-			case codes.PermissionDenied:
-				w.WriteHeader(http.StatusForbidden)
-				h.logger.Info("Failed to write to GRPC resource: User not authenticated", "error", err)
-			case codes.Internal:
-				w.WriteHeader(http.StatusInternalServerError)
-				h.logger.Error("Failed to write to GRPC resource: Internal error", "error", err)
-			case codes.Aborted:
-				w.WriteHeader(http.StatusInternalServerError)
-				h.logger.Error("Failed to write to GRPC resource: GRPC aborted", "error", err)
-			default:
-				w.WriteHeader(http.StatusInternalServerError)
-				h.logger.Error("Failed to write to GRPC resource", "error", err)
-			}
-		} else {
-			w.WriteHeader(http.StatusInternalServerError)
-			h.logger.Error("Failed to write to GRPC resource: not able to parse error returned", "error", err)
-		}
+		handleResponseError(err, w, h)
 		return
 	}
 
@@ -170,3 +152,22 @@ func jsonMarshal(res *pbresource.Resource) ([]byte, error) {
 	delete(stuff["data"].(map[string]any), "@type")
 	return json.MarshalIndent(stuff, "", "  ")
 }
+
+func handleResponseError(err error, w http.ResponseWriter, h *resourceHandler) {
+	if e, ok := status.FromError(err); ok {
+		switch e.Code() {
+		case codes.PermissionDenied:
+			w.WriteHeader(http.StatusForbidden)
+			h.logger.Info("Failed to write to GRPC resource: User not authenticated", "error", err)
+		case codes.NotFound:
+			w.WriteHeader(http.StatusNotFound)
+			h.logger.Info("Failed to write to GRPC resource: Not found", "error", err)
+		default:
+			w.WriteHeader(http.StatusInternalServerError)
+			h.logger.Error("Failed to write to GRPC resource", "error", err)
+		}
+	} else {
+		w.WriteHeader(http.StatusInternalServerError)
+		h.logger.Error("Failed to write to GRPC resource: not able to parse error returned", "error", err)
+	}
+}

From 1c2efc21aef7ad279eaefc10ceca45262f7da535 Mon Sep 17 00:00:00 2001
From: Xinyi Wang <xinyi.wang@hashicorp.com>
Date: Tue, 25 Jul 2023 12:16:37 -0700
Subject: [PATCH 13/29] add more tests

---
 internal/resource/http/http.go      |  3 ++
 internal/resource/http/http_test.go | 43 +++++++++++++++++++++++++++--
 2 files changed, 44 insertions(+), 2 deletions(-)

diff --git a/internal/resource/http/http.go b/internal/resource/http/http.go
index 7a988eb7961b..f2fb3de5ae48 100644
--- a/internal/resource/http/http.go
+++ b/internal/resource/http/http.go
@@ -134,6 +134,9 @@ func checkURL(r *http.Request) (tenancy *pbresource.Tenancy, resourceName string
 		}
 	}
 	resourceName = path.Base(r.URL.Path)
+	if resourceName == "." || resourceName == "/" {
+		resourceName = ""
+	}
 
 	return
 }
diff --git a/internal/resource/http/http_test.go b/internal/resource/http/http_test.go
index 6bc9be9a2cf0..168c1d5d2a0e 100644
--- a/internal/resource/http/http_test.go
+++ b/internal/resource/http/http_test.go
@@ -31,9 +31,47 @@ func TestResourceHandler(t *testing.T) {
 		hclog.NewNullLogger(),
 	}
 
-	t.Run("Write", func(t *testing.T) {
+	t.Run("should return bad request due to missing resource name", func(t *testing.T) {
 		rsp := httptest.NewRecorder()
-		req := httptest.NewRequest("PUT", "/api/demo/v2/artist/keith-urban?partition=default&peer_name=local&namespace=default", strings.NewReader(`
+		req := httptest.NewRequest("PUT", "/?partition=default&peer_name=local&namespace=default", strings.NewReader(`
+			{
+				"metadata": {
+					"foo": "bar"
+				},
+				"data": {
+					"name": "Keith Urban",
+					"genre": "GENRE_COUNTRY"
+				}
+			}
+		`))
+
+		resourceHandler.ServeHTTP(rsp, req)
+
+		require.Equal(t, http.StatusBadRequest, rsp.Result().StatusCode)
+	})
+
+	t.Run("should return bad request due to wrong schema", func(t *testing.T) {
+		rsp := httptest.NewRecorder()
+		req := httptest.NewRequest("PUT", "/?partition=default&peer_name=local&namespace=default", strings.NewReader(`
+			{
+				"metadata": {
+					"foo": "bar"
+				},
+				"tada": {
+					"name": "Keith Urban",
+					"genre": "GENRE_COUNTRY"
+				}
+			}
+		`))
+
+		resourceHandler.ServeHTTP(rsp, req)
+
+		require.Equal(t, http.StatusBadRequest, rsp.Result().StatusCode)
+	})
+
+	t.Run("should write to the resource backend", func(t *testing.T) {
+		rsp := httptest.NewRecorder()
+		req := httptest.NewRequest("PUT", "/keith-urban?partition=default&peer_name=local&namespace=default", strings.NewReader(`
 			{
 				"metadata": {
 					"foo": "bar"
@@ -52,6 +90,7 @@ func TestResourceHandler(t *testing.T) {
 		var result map[string]any
 		require.NoError(t, json.NewDecoder(rsp.Body).Decode(&result))
 		require.Equal(t, "Keith Urban", result["data"].(map[string]any)["name"])
+		require.Equal(t, "keith-urban", result["id"].(map[string]any)["name"])
 
 		readRsp, err := client.Read(testutil.TestContext(t), &pbresource.ReadRequest{
 			Id: &pbresource.ID{

From dacbda0183e58e2a1c4e2e92c2b8889c3631cfdb Mon Sep 17 00:00:00 2001
From: Xinyi Wang <xinyi.wang@hashicorp.com>
Date: Thu, 27 Jul 2023 12:19:03 -0700
Subject: [PATCH 14/29] refactor test

---
 .../services/resource/testing/testing.go      |  66 +++++++++++
 internal/resource/http/http_test.go           | 106 ++++++++++++------
 2 files changed, 139 insertions(+), 33 deletions(-)

diff --git a/agent/grpc-external/services/resource/testing/testing.go b/agent/grpc-external/services/resource/testing/testing.go
index 5bcbc148e7bb..b315f9405faa 100644
--- a/agent/grpc-external/services/resource/testing/testing.go
+++ b/agent/grpc-external/services/resource/testing/testing.go
@@ -8,15 +8,42 @@ import (
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
 
+	"github.com/hashicorp/consul/acl"
 	"github.com/hashicorp/consul/acl/resolver"
 	svc "github.com/hashicorp/consul/agent/grpc-external/services/resource"
 	internal "github.com/hashicorp/consul/agent/grpc-internal"
+	"github.com/hashicorp/consul/agent/structs"
 	"github.com/hashicorp/consul/internal/resource"
 	"github.com/hashicorp/consul/internal/storage/inmem"
 	"github.com/hashicorp/consul/proto-public/pbresource"
 	"github.com/hashicorp/consul/sdk/testutil"
+	"github.com/hashicorp/go-uuid"
 )
 
+func randomACLIdentity(t *testing.T) structs.ACLIdentity {
+	id, err := uuid.GenerateUUID()
+	require.NoError(t, err)
+
+	return &structs.ACLToken{AccessorID: id}
+}
+
+func AuthorizerFrom(t *testing.T, policyStrs ...string) resolver.Result {
+	policies := []*acl.Policy{}
+	for _, policyStr := range policyStrs {
+		policy, err := acl.NewPolicyFromSource(policyStr, nil, nil)
+		require.NoError(t, err)
+		policies = append(policies, policy)
+	}
+
+	authz, err := acl.NewPolicyAuthorizerWithDefaults(acl.DenyAll(), policies, nil)
+	require.NoError(t, err)
+
+	return resolver.Result{
+		Authorizer:  authz,
+		ACLIdentity: randomACLIdentity(t),
+	}
+}
+
 // RunResourceService runs a Resource Service for the duration of the test and
 // returns a client to interact with it. ACLs will be disabled.
 func RunResourceService(t *testing.T, registerFns ...func(resource.Registry)) pbresource.ResourceServiceClient {
@@ -57,3 +84,42 @@ func RunResourceService(t *testing.T, registerFns ...func(resource.Registry)) pb
 
 	return pbresource.NewResourceServiceClient(conn)
 }
+
+func RunResourceServiceWithACL(t *testing.T, aclResolver svc.ACLResolver, registerFns ...func(resource.Registry)) pbresource.ResourceServiceClient {
+	t.Helper()
+
+	backend, err := inmem.NewBackend()
+	require.NoError(t, err)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	t.Cleanup(cancel)
+	go backend.Run(ctx)
+
+	registry := resource.NewRegistry()
+	for _, fn := range registerFns {
+		fn(registry)
+	}
+
+	server := grpc.NewServer()
+
+	svc.NewServer(svc.Config{
+		Backend:     backend,
+		Registry:    registry,
+		Logger:      testutil.Logger(t),
+		ACLResolver: aclResolver,
+	}).Register(server)
+
+	pipe := internal.NewPipeListener()
+	go server.Serve(pipe)
+	t.Cleanup(server.Stop)
+
+	conn, err := grpc.Dial("",
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+		grpc.WithContextDialer(pipe.DialContext),
+		grpc.WithBlock(),
+	)
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = conn.Close() })
+
+	return pbresource.NewResourceServiceClient(conn)
+}
diff --git a/internal/resource/http/http_test.go b/internal/resource/http/http_test.go
index 168c1d5d2a0e..ae152135ca4c 100644
--- a/internal/resource/http/http_test.go
+++ b/internal/resource/http/http_test.go
@@ -8,9 +8,13 @@ import (
 	"testing"
 
 	"github.com/hashicorp/go-hclog"
+	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
 
+	"github.com/hashicorp/consul/acl"
+	resourceSvc "github.com/hashicorp/consul/agent/grpc-external/services/resource"
 	svctest "github.com/hashicorp/consul/agent/grpc-external/services/resource/testing"
+
 	"github.com/hashicorp/consul/internal/resource"
 	"github.com/hashicorp/consul/internal/resource/demo"
 	"github.com/hashicorp/consul/proto-public/pbresource"
@@ -18,9 +22,20 @@ import (
 	"github.com/hashicorp/consul/sdk/testutil"
 )
 
-func TestResourceHandler(t *testing.T) {
-	client := svctest.RunResourceService(t, demo.RegisterTypes)
+const testACLToken = acl.AnonymousTokenID
+
+func parseToken(req *http.Request, token *string) {
+	*token = req.Header.Get("x-Consul-Token")
+}
 
+func TestResourceHandler_InputValidation(t *testing.T) {
+	type testCase struct {
+		description          string
+		request              *http.Request
+		response             *httptest.ResponseRecorder
+		expectedResponseCode int
+	}
+	client := svctest.RunResourceService(t, demo.RegisterTypes)
 	resourceHandler := resourceHandler{
 		resource.Registration{
 			Type:  demo.TypeV2Artist,
@@ -31,43 +46,66 @@ func TestResourceHandler(t *testing.T) {
 		hclog.NewNullLogger(),
 	}
 
-	t.Run("should return bad request due to missing resource name", func(t *testing.T) {
-		rsp := httptest.NewRecorder()
-		req := httptest.NewRequest("PUT", "/?partition=default&peer_name=local&namespace=default", strings.NewReader(`
-			{
-				"metadata": {
-					"foo": "bar"
-				},
-				"data": {
-					"name": "Keith Urban",
-					"genre": "GENRE_COUNTRY"
+	testCases := []testCase{
+		{
+			description: "missing resource name",
+			request: httptest.NewRequest("PUT", "/?partition=default&peer_name=local&namespace=default", strings.NewReader(`
+				{
+					"metadata": {
+						"foo": "bar"
+					},
+					"data": {
+						"name": "Keith Urban",
+						"genre": "GENRE_COUNTRY"
+					}
 				}
-			}
-		`))
+			`)),
+			response:             httptest.NewRecorder(),
+			expectedResponseCode: http.StatusBadRequest,
+		},
+		{
+			description: "wrong schema",
+			request: httptest.NewRequest("PUT", "/?partition=default&peer_name=local&namespace=default", strings.NewReader(`
+				{
+					"metadata": {
+						"foo": "bar"
+					},
+					"tada": {
+						"name": "Keith Urban",
+						"genre": "GENRE_COUNTRY"
+					}
+				}
+			`)),
+			response:             httptest.NewRecorder(),
+			expectedResponseCode: http.StatusBadRequest,
+		},
+	}
 
-		resourceHandler.ServeHTTP(rsp, req)
+	for _, tc := range testCases {
+		t.Run(tc.description, func(t *testing.T) {
+			resourceHandler.ServeHTTP(tc.response, tc.request)
 
-		require.Equal(t, http.StatusBadRequest, rsp.Result().StatusCode)
-	})
+			require.Equal(t, tc.expectedResponseCode, tc.response.Result().StatusCode)
+		})
+	}
+}
 
-	t.Run("should return bad request due to wrong schema", func(t *testing.T) {
-		rsp := httptest.NewRecorder()
-		req := httptest.NewRequest("PUT", "/?partition=default&peer_name=local&namespace=default", strings.NewReader(`
-			{
-				"metadata": {
-					"foo": "bar"
-				},
-				"tada": {
-					"name": "Keith Urban",
-					"genre": "GENRE_COUNTRY"
-				}
-			}
-		`))
+func TestResourceWriteHandler(t *testing.T) {
+	aclResolver := &resourceSvc.MockACLResolver{}
+	aclResolver.On("ResolveTokenAndDefaultMeta", testACLToken, mock.Anything, mock.Anything).
+		Return(svctest.AuthorizerFrom(t, demo.ArtistV2WritePolicy), nil)
 
-		resourceHandler.ServeHTTP(rsp, req)
+	client := svctest.RunResourceServiceWithACL(t, aclResolver, demo.RegisterTypes)
 
-		require.Equal(t, http.StatusBadRequest, rsp.Result().StatusCode)
-	})
+	resourceHandler := resourceHandler{
+		resource.Registration{
+			Type:  demo.TypeV2Artist,
+			Proto: &pbdemov2.Artist{},
+		},
+		client,
+		parseToken,
+		hclog.NewNullLogger(),
+	}
 
 	t.Run("should write to the resource backend", func(t *testing.T) {
 		rsp := httptest.NewRecorder()
@@ -83,6 +121,8 @@ func TestResourceHandler(t *testing.T) {
 			}
 		`))
 
+		req.Header.Add("x-consul-token", testACLToken)
+
 		resourceHandler.ServeHTTP(rsp, req)
 
 		require.Equal(t, http.StatusOK, rsp.Result().StatusCode)

From b5edf4a4eeea65383ac0198680ed00eb7e409e0f Mon Sep 17 00:00:00 2001
From: Xinyi Wang <xinyi.wang@hashicorp.com>
Date: Thu, 27 Jul 2023 16:10:12 -0700
Subject: [PATCH 15/29] add add owner and remove extra check

---
 internal/resource/http/http.go | 31 +++++++++----------------------
 1 file changed, 9 insertions(+), 22 deletions(-)

diff --git a/internal/resource/http/http.go b/internal/resource/http/http.go
index f2fb3de5ae48..2ae3fd24357f 100644
--- a/internal/resource/http/http.go
+++ b/internal/resource/http/http.go
@@ -37,10 +37,10 @@ func NewHandler(
 }
 
 type writeRequest struct {
-	// TODO: Owner.
 	Version  string            `json:"version"`
 	Metadata map[string]string `json:"metadata"`
 	Data     json.RawMessage   `json:"data"`
+	Owner    *pbresource.ID    `json:"owner"`
 }
 
 type resourceHandler struct {
@@ -65,12 +65,12 @@ func (h *resourceHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 
 func (h *resourceHandler) handleWrite(w http.ResponseWriter, r *http.Request, ctx context.Context) {
 	var req writeRequest
-	// convert req data to struct
+	// convert req body to writeRequest
 	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
 		w.WriteHeader(http.StatusBadRequest)
 		w.Write([]byte("Request body didn't follow schema."))
 	}
-	// struct to proto message
+	// convert data struct to proto message
 	data := h.reg.Proto.ProtoReflect().New().Interface()
 	if err := protojson.Unmarshal(req.Data, data); err != nil {
 		w.WriteHeader(http.StatusBadRequest)
@@ -85,14 +85,7 @@ func (h *resourceHandler) handleWrite(w http.ResponseWriter, r *http.Request, ct
 	}
 
 	tenancyInfo, resourceName := checkURL(r)
-	if tenancyInfo == nil {
-		w.WriteHeader(http.StatusBadRequest)
-		w.Write([]byte("Query params partition, peer_name, and namespace are required."))
-	}
-	if resourceName == "" {
-		w.WriteHeader(http.StatusBadRequest)
-		w.Write([]byte("Missing resource name in the URL"))
-	}
+
 	rsp, err := h.client.Write(ctx, &pbresource.WriteRequest{
 		Resource: &pbresource.Resource{
 			Id: &pbresource.ID{
@@ -100,6 +93,7 @@ func (h *resourceHandler) handleWrite(w http.ResponseWriter, r *http.Request, ct
 				Tenancy: tenancyInfo,
 				Name:    resourceName,
 			},
+			Owner:    req.Owner,
 			Version:  req.Version,
 			Metadata: req.Metadata,
 			Data:     anyProtoMsg,
@@ -121,17 +115,10 @@ func (h *resourceHandler) handleWrite(w http.ResponseWriter, r *http.Request, ct
 
 func checkURL(r *http.Request) (tenancy *pbresource.Tenancy, resourceName string) {
 	params := r.URL.Query()
-	partition := params.Get("partition")
-	peerName := params.Get("peer_name")
-	namespace := params.Get("namespace")
-	if partition == "" || peerName == "" || namespace == "" {
-		tenancy = nil
-	} else {
-		tenancy = &pbresource.Tenancy{
-			Partition: partition,
-			PeerName:  peerName,
-			Namespace: namespace,
-		}
+	tenancy = &pbresource.Tenancy{
+		Partition: params.Get("partition"),
+		PeerName:  params.Get("peer_name"),
+		Namespace: params.Get("namespace"),
 	}
 	resourceName = path.Base(r.URL.Path)
 	if resourceName == "." || resourceName == "/" {

From 8db104c3031217d98c23241a10b7988ca4bf87f5 Mon Sep 17 00:00:00 2001
From: Xinyi Wang <xinyi.wang@hashicorp.com>
Date: Wed, 2 Aug 2023 15:23:35 -0700
Subject: [PATCH 16/29] add more tests

---
 internal/resource/http/http.go      | 17 +++++-----
 internal/resource/http/http_test.go | 49 ++++++++++++++++++++++++++---
 2 files changed, 55 insertions(+), 11 deletions(-)

diff --git a/internal/resource/http/http.go b/internal/resource/http/http.go
index 2ae3fd24357f..76d2e63e43a1 100644
--- a/internal/resource/http/http.go
+++ b/internal/resource/http/http.go
@@ -37,7 +37,6 @@ func NewHandler(
 }
 
 type writeRequest struct {
-	Version  string            `json:"version"`
 	Metadata map[string]string `json:"metadata"`
 	Data     json.RawMessage   `json:"data"`
 	Owner    *pbresource.ID    `json:"owner"`
@@ -84,7 +83,7 @@ func (h *resourceHandler) handleWrite(w http.ResponseWriter, r *http.Request, ct
 		return
 	}
 
-	tenancyInfo, resourceName := checkURL(r)
+	tenancyInfo, resourceName, version := checkURL(r)
 
 	rsp, err := h.client.Write(ctx, &pbresource.WriteRequest{
 		Resource: &pbresource.Resource{
@@ -94,7 +93,7 @@ func (h *resourceHandler) handleWrite(w http.ResponseWriter, r *http.Request, ct
 				Name:    resourceName,
 			},
 			Owner:    req.Owner,
-			Version:  req.Version,
+			Version:  version,
 			Metadata: req.Metadata,
 			Data:     anyProtoMsg,
 		},
@@ -113,7 +112,7 @@ func (h *resourceHandler) handleWrite(w http.ResponseWriter, r *http.Request, ct
 	w.Write(output)
 }
 
-func checkURL(r *http.Request) (tenancy *pbresource.Tenancy, resourceName string) {
+func checkURL(r *http.Request) (tenancy *pbresource.Tenancy, resourceName string, version string) {
 	params := r.URL.Query()
 	tenancy = &pbresource.Tenancy{
 		Partition: params.Get("partition"),
@@ -124,6 +123,7 @@ func checkURL(r *http.Request) (tenancy *pbresource.Tenancy, resourceName string
 	if resourceName == "." || resourceName == "/" {
 		resourceName = ""
 	}
+	version = params.Get("version")
 
 	return
 }
@@ -146,12 +146,15 @@ func jsonMarshal(res *pbresource.Resource) ([]byte, error) {
 func handleResponseError(err error, w http.ResponseWriter, h *resourceHandler) {
 	if e, ok := status.FromError(err); ok {
 		switch e.Code() {
-		case codes.PermissionDenied:
-			w.WriteHeader(http.StatusForbidden)
-			h.logger.Info("Failed to write to GRPC resource: User not authenticated", "error", err)
+		case codes.InvalidArgument:
+			w.WriteHeader(http.StatusBadRequest)
+			h.logger.Info("User has mal-formed request", "error", err)
 		case codes.NotFound:
 			w.WriteHeader(http.StatusNotFound)
 			h.logger.Info("Failed to write to GRPC resource: Not found", "error", err)
+		case codes.PermissionDenied:
+			w.WriteHeader(http.StatusForbidden)
+			h.logger.Info("Failed to write to GRPC resource: User not authenticated", "error", err)
 		default:
 			w.WriteHeader(http.StatusInternalServerError)
 			h.logger.Error("Failed to write to GRPC resource", "error", err)
diff --git a/internal/resource/http/http_test.go b/internal/resource/http/http_test.go
index ae152135ca4c..0d0bad5da219 100644
--- a/internal/resource/http/http_test.go
+++ b/internal/resource/http/http_test.go
@@ -22,7 +22,8 @@ import (
 	"github.com/hashicorp/consul/sdk/testutil"
 )
 
-const testACLToken = acl.AnonymousTokenID
+const testACLTokenArtistV2WritePolicy = acl.AnonymousTokenID
+const testACLTokenArtistV2ReadPolicy = "00000000-0000-0000-0000-000000000001"
 
 func parseToken(req *http.Request, token *string) {
 	*token = req.Header.Get("x-Consul-Token")
@@ -67,10 +68,27 @@ func TestResourceHandler_InputValidation(t *testing.T) {
 			description: "wrong schema",
 			request: httptest.NewRequest("PUT", "/?partition=default&peer_name=local&namespace=default", strings.NewReader(`
 				{
+					"version": "test_version",
 					"metadata": {
 						"foo": "bar"
 					},
-					"tada": {
+					"data": {
+						"name": "Keith Urban",
+						"genre": "GENRE_COUNTRY"
+					}
+				}
+			`)),
+			response:             httptest.NewRecorder(),
+			expectedResponseCode: http.StatusBadRequest,
+		},
+		{
+			description: "missing tenancy info",
+			request: httptest.NewRequest("PUT", "/keith-urban?partition=default&peer_name=local", strings.NewReader(`
+				{
+					"metadata": {
+						"foo": "bar"
+					},
+					"data": {
 						"name": "Keith Urban",
 						"genre": "GENRE_COUNTRY"
 					}
@@ -92,8 +110,10 @@ func TestResourceHandler_InputValidation(t *testing.T) {
 
 func TestResourceWriteHandler(t *testing.T) {
 	aclResolver := &resourceSvc.MockACLResolver{}
-	aclResolver.On("ResolveTokenAndDefaultMeta", testACLToken, mock.Anything, mock.Anything).
+	aclResolver.On("ResolveTokenAndDefaultMeta", testACLTokenArtistV2WritePolicy, mock.Anything, mock.Anything).
 		Return(svctest.AuthorizerFrom(t, demo.ArtistV2WritePolicy), nil)
+	aclResolver.On("ResolveTokenAndDefaultMeta", testACLTokenArtistV2ReadPolicy, mock.Anything, mock.Anything).
+		Return(svctest.AuthorizerFrom(t, demo.ArtistV2ReadPolicy), nil)
 
 	client := svctest.RunResourceServiceWithACL(t, aclResolver, demo.RegisterTypes)
 
@@ -107,6 +127,27 @@ func TestResourceWriteHandler(t *testing.T) {
 		hclog.NewNullLogger(),
 	}
 
+	t.Run("should be blocked if the token is not authorized", func(t *testing.T) {
+		rsp := httptest.NewRecorder()
+		req := httptest.NewRequest("PUT", "/keith-urban?partition=default&peer_name=local&namespace=default", strings.NewReader(`
+			{
+				"metadata": {
+					"foo": "bar"
+				},
+				"data": {
+					"name": "Keith Urban",
+					"genre": "GENRE_COUNTRY"
+				}
+			}
+		`))
+
+		req.Header.Add("x-consul-token", testACLTokenArtistV2ReadPolicy)
+
+		resourceHandler.ServeHTTP(rsp, req)
+
+		require.Equal(t, http.StatusForbidden, rsp.Result().StatusCode)
+	})
+
 	t.Run("should write to the resource backend", func(t *testing.T) {
 		rsp := httptest.NewRecorder()
 		req := httptest.NewRequest("PUT", "/keith-urban?partition=default&peer_name=local&namespace=default", strings.NewReader(`
@@ -121,7 +162,7 @@ func TestResourceWriteHandler(t *testing.T) {
 			}
 		`))
 
-		req.Header.Add("x-consul-token", testACLToken)
+		req.Header.Add("x-consul-token", testACLTokenArtistV2WritePolicy)
 
 		resourceHandler.ServeHTTP(rsp, req)
 

From f326d53aecd0a978b817e70670b21cd1b7047ac3 Mon Sep 17 00:00:00 2001
From: Xinyi Wang <xinyi.wang@hashicorp.com>
Date: Thu, 3 Aug 2023 09:58:24 -0700
Subject: [PATCH 17/29] refactor checkUrl function

---
 internal/resource/http/http.go | 34 ++++++++++++++++++----------------
 1 file changed, 18 insertions(+), 16 deletions(-)

diff --git a/internal/resource/http/http.go b/internal/resource/http/http.go
index e054138e339b..34da16685e83 100644
--- a/internal/resource/http/http.go
+++ b/internal/resource/http/http.go
@@ -85,17 +85,17 @@ func (h *resourceHandler) handleWrite(w http.ResponseWriter, r *http.Request, ct
 		return
 	}
 
-	tenancyInfo, resourceName, version := checkURL(r)
+	tenancyInfo, params := checkURL(r)
 
 	rsp, err := h.client.Write(ctx, &pbresource.WriteRequest{
 		Resource: &pbresource.Resource{
 			Id: &pbresource.ID{
 				Type:    h.reg.Type,
 				Tenancy: tenancyInfo,
-				Name:    resourceName,
+				Name:    params["resourceName"],
 			},
 			Owner:    req.Owner,
-			Version:  version,
+			Version:  params["version"],
 			Metadata: req.Metadata,
 			Data:     anyProtoMsg,
 		},
@@ -115,20 +115,17 @@ func (h *resourceHandler) handleWrite(w http.ResponseWriter, r *http.Request, ct
 }
 
 func (h *resourceHandler) handleRead(w http.ResponseWriter, r *http.Request, ctx context.Context) {
-	tenancyInfo, resourceName, _ := checkURL(r)
+	tenancyInfo, params := checkURL(r)
 	if tenancyInfo == nil {
 		w.WriteHeader(http.StatusBadRequest)
 		w.Write([]byte("Missing partition, peer_name or namespace in the query params"))
 	}
-	if resourceName == "" {
-		w.WriteHeader(http.StatusBadRequest)
-		w.Write([]byte("Missing resource name in the URL"))
-	}
+
 	rsp, err := h.client.Read(ctx, &pbresource.ReadRequest{
 		Id: &pbresource.ID{
 			Type:    h.reg.Type,
 			Tenancy: tenancyInfo,
-			Name:    resourceName,
+			Name:    params["resourceName"],
 		},
 	})
 	if err != nil {
@@ -145,18 +142,23 @@ func (h *resourceHandler) handleRead(w http.ResponseWriter, r *http.Request, ctx
 	w.Write(output)
 }
 
-func checkURL(r *http.Request) (tenancy *pbresource.Tenancy, resourceName string, version string) {
-	params := r.URL.Query()
+func checkURL(r *http.Request) (tenancy *pbresource.Tenancy, params map[string]string) {
+	query := r.URL.Query()
 	tenancy = &pbresource.Tenancy{
-		Partition: params.Get("partition"),
-		PeerName:  params.Get("peer_name"),
-		Namespace: params.Get("namespace"),
+		Partition: query.Get("partition"),
+		PeerName:  query.Get("peer_name"),
+		Namespace: query.Get("namespace"),
 	}
-	resourceName = path.Base(r.URL.Path)
+
+	resourceName := path.Base(r.URL.Path)
 	if resourceName == "." || resourceName == "/" {
 		resourceName = ""
 	}
-	version = params.Get("version")
+
+	params = make(map[string]string)
+	params["resourceName"] = resourceName
+	params["version"] = query.Get("version")
+	params["consistent"] = query.Get("consistent")
 
 	return
 }

From 388a4656c02a89224876075d82b274f6b3ba853a Mon Sep 17 00:00:00 2001
From: Xinyi Wang <xinyi.wang@hashicorp.com>
Date: Thu, 3 Aug 2023 11:43:52 -0700
Subject: [PATCH 18/29] write the error msg back

---
 internal/resource/http/http.go | 1 +
 1 file changed, 1 insertion(+)

diff --git a/internal/resource/http/http.go b/internal/resource/http/http.go
index 76d2e63e43a1..45d8d6c2773e 100644
--- a/internal/resource/http/http.go
+++ b/internal/resource/http/http.go
@@ -163,4 +163,5 @@ func handleResponseError(err error, w http.ResponseWriter, h *resourceHandler) {
 		w.WriteHeader(http.StatusInternalServerError)
 		h.logger.Error("Failed to write to GRPC resource: not able to parse error returned", "error", err)
 	}
+	w.Write([]byte(err.Error()))
 }

From c3775a765d5a190109dd0958a7db7a55c5f0966f Mon Sep 17 00:00:00 2001
From: Xinyi Wang <xinyi.wang@hashicorp.com>
Date: Thu, 3 Aug 2023 12:05:47 -0700
Subject: [PATCH 19/29] add owner test

---
 internal/resource/http/http_test.go | 92 ++++++++++++++++++++++++-----
 1 file changed, 78 insertions(+), 14 deletions(-)

diff --git a/internal/resource/http/http_test.go b/internal/resource/http/http_test.go
index 0d0bad5da219..96da8d1cdb9e 100644
--- a/internal/resource/http/http_test.go
+++ b/internal/resource/http/http_test.go
@@ -11,9 +11,9 @@ import (
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
 
-	"github.com/hashicorp/consul/acl"
 	resourceSvc "github.com/hashicorp/consul/agent/grpc-external/services/resource"
 	svctest "github.com/hashicorp/consul/agent/grpc-external/services/resource/testing"
+	pbdemov1 "github.com/hashicorp/consul/proto/private/pbdemo/v1"
 
 	"github.com/hashicorp/consul/internal/resource"
 	"github.com/hashicorp/consul/internal/resource/demo"
@@ -22,8 +22,8 @@ import (
 	"github.com/hashicorp/consul/sdk/testutil"
 )
 
-const testACLTokenArtistV2WritePolicy = acl.AnonymousTokenID
-const testACLTokenArtistV2ReadPolicy = "00000000-0000-0000-0000-000000000001"
+const testACLTokenArtistReadPolicy = "00000000-0000-0000-0000-000000000001"
+const testACLTokenArtistWritePolicy = "00000000-0000-0000-0000-000000000002"
 
 func parseToken(req *http.Request, token *string) {
 	*token = req.Header.Get("x-Consul-Token")
@@ -110,14 +110,24 @@ func TestResourceHandler_InputValidation(t *testing.T) {
 
 func TestResourceWriteHandler(t *testing.T) {
 	aclResolver := &resourceSvc.MockACLResolver{}
-	aclResolver.On("ResolveTokenAndDefaultMeta", testACLTokenArtistV2WritePolicy, mock.Anything, mock.Anything).
-		Return(svctest.AuthorizerFrom(t, demo.ArtistV2WritePolicy), nil)
-	aclResolver.On("ResolveTokenAndDefaultMeta", testACLTokenArtistV2ReadPolicy, mock.Anything, mock.Anything).
-		Return(svctest.AuthorizerFrom(t, demo.ArtistV2ReadPolicy), nil)
+	aclResolver.On("ResolveTokenAndDefaultMeta", testACLTokenArtistReadPolicy, mock.Anything, mock.Anything).
+		Return(svctest.AuthorizerFrom(t, demo.ArtistV1ReadPolicy, demo.ArtistV2ReadPolicy), nil)
+	aclResolver.On("ResolveTokenAndDefaultMeta", testACLTokenArtistWritePolicy, mock.Anything, mock.Anything).
+		Return(svctest.AuthorizerFrom(t, demo.ArtistV1WritePolicy, demo.ArtistV2WritePolicy), nil)
 
 	client := svctest.RunResourceServiceWithACL(t, aclResolver, demo.RegisterTypes)
 
-	resourceHandler := resourceHandler{
+	v1ArtistHandler := resourceHandler{
+		resource.Registration{
+			Type:  demo.TypeV1Artist,
+			Proto: &pbdemov1.Artist{},
+		},
+		client,
+		parseToken,
+		hclog.NewNullLogger(),
+	}
+
+	v2ArtistHandler := resourceHandler{
 		resource.Registration{
 			Type:  demo.TypeV2Artist,
 			Proto: &pbdemov2.Artist{},
@@ -129,7 +139,7 @@ func TestResourceWriteHandler(t *testing.T) {
 
 	t.Run("should be blocked if the token is not authorized", func(t *testing.T) {
 		rsp := httptest.NewRecorder()
-		req := httptest.NewRequest("PUT", "/keith-urban?partition=default&peer_name=local&namespace=default", strings.NewReader(`
+		req := httptest.NewRequest("PUT", "/demo/v2/artist/keith-urban?partition=default&peer_name=local&namespace=default", strings.NewReader(`
 			{
 				"metadata": {
 					"foo": "bar"
@@ -141,16 +151,16 @@ func TestResourceWriteHandler(t *testing.T) {
 			}
 		`))
 
-		req.Header.Add("x-consul-token", testACLTokenArtistV2ReadPolicy)
+		req.Header.Add("x-consul-token", testACLTokenArtistReadPolicy)
 
-		resourceHandler.ServeHTTP(rsp, req)
+		v2ArtistHandler.ServeHTTP(rsp, req)
 
 		require.Equal(t, http.StatusForbidden, rsp.Result().StatusCode)
 	})
 
 	t.Run("should write to the resource backend", func(t *testing.T) {
 		rsp := httptest.NewRecorder()
-		req := httptest.NewRequest("PUT", "/keith-urban?partition=default&peer_name=local&namespace=default", strings.NewReader(`
+		req := httptest.NewRequest("PUT", "/demo/v2/artist/keith-urban?partition=default&peer_name=local&namespace=default", strings.NewReader(`
 			{
 				"metadata": {
 					"foo": "bar"
@@ -162,9 +172,9 @@ func TestResourceWriteHandler(t *testing.T) {
 			}
 		`))
 
-		req.Header.Add("x-consul-token", testACLTokenArtistV2WritePolicy)
+		req.Header.Add("x-consul-token", testACLTokenArtistWritePolicy)
 
-		resourceHandler.ServeHTTP(rsp, req)
+		v2ArtistHandler.ServeHTTP(rsp, req)
 
 		require.Equal(t, http.StatusOK, rsp.Result().StatusCode)
 
@@ -187,4 +197,58 @@ func TestResourceWriteHandler(t *testing.T) {
 		require.NoError(t, readRsp.Resource.Data.UnmarshalTo(&artist))
 		require.Equal(t, "Keith Urban", artist.Name)
 	})
+
+	t.Run("should write to the resource backend with owner", func(t *testing.T) {
+		rsp := httptest.NewRecorder()
+		req := httptest.NewRequest("PUT", "/demo/v1/artist/keith-urban-v1?partition=default&peer_name=local&namespace=default", strings.NewReader(`
+			{
+				"metadata": {
+					"foo": "bar"
+				},
+				"data": {
+					"name": "Keith Urban V1",
+					"genre": "GENRE_COUNTRY"
+				},
+				"owner": {
+					"name": "keith-urban",
+					"type": {
+						"group": "demo",
+						"group_version": "v2",
+						"kind": "Artist"
+					},
+					"tenancy": {
+						"partition": "default",
+						"peer_name": "local",
+						"namespace": "default"
+					}
+				}
+			}
+		`))
+
+		req.Header.Add("x-consul-token", testACLTokenArtistWritePolicy)
+
+		v1ArtistHandler.ServeHTTP(rsp, req)
+
+		require.Equal(t, http.StatusOK, rsp.Result().StatusCode)
+
+		var result map[string]any
+		require.NoError(t, json.NewDecoder(rsp.Body).Decode(&result))
+		require.Equal(t, "Keith Urban V1", result["data"].(map[string]any)["name"])
+		require.Equal(t, "keith-urban-v1", result["id"].(map[string]any)["name"])
+
+		readRsp, err := client.Read(testutil.TestContext(t), &pbresource.ReadRequest{
+			Id: &pbresource.ID{
+				Type:    demo.TypeV1Artist,
+				Tenancy: demo.TenancyDefault,
+				Name:    "keith-urban-v1",
+			},
+		})
+		require.NoError(t, err)
+		require.NotNil(t, readRsp.Resource)
+		require.Equal(t, "keith-urban", readRsp.Resource.Owner.Name)
+
+		var artist pbdemov1.Artist
+		require.NoError(t, readRsp.Resource.Data.UnmarshalTo(&artist))
+		require.Equal(t, "Keith Urban V1", artist.Name)
+	})
 }

From 7e2c9faeeaf6924cf89106c669770976ef92d9d4 Mon Sep 17 00:00:00 2001
From: Xinyi Wang <xinyi.wang@hashicorp.com>
Date: Thu, 3 Aug 2023 12:35:12 -0700
Subject: [PATCH 20/29] could use the consistent mode

---
 internal/resource/http/http.go      | 9 +++++----
 internal/resource/http/http_test.go | 2 +-
 2 files changed, 6 insertions(+), 5 deletions(-)

diff --git a/internal/resource/http/http.go b/internal/resource/http/http.go
index 0cd7111b120e..5a4fb3d844de 100644
--- a/internal/resource/http/http.go
+++ b/internal/resource/http/http.go
@@ -116,9 +116,8 @@ func (h *resourceHandler) handleWrite(w http.ResponseWriter, r *http.Request, ct
 
 func (h *resourceHandler) handleRead(w http.ResponseWriter, r *http.Request, ctx context.Context) {
 	tenancyInfo, params := checkURL(r)
-	if tenancyInfo == nil {
-		w.WriteHeader(http.StatusBadRequest)
-		w.Write([]byte("Missing partition, peer_name or namespace in the query params"))
+	if params["consistent"] != "" {
+		ctx = metadata.AppendToOutgoingContext(ctx, "x-consul-consistency-mode", "consistent")
 	}
 
 	rsp, err := h.client.Read(ctx, &pbresource.ReadRequest{
@@ -158,7 +157,9 @@ func checkURL(r *http.Request) (tenancy *pbresource.Tenancy, params map[string]s
 	params = make(map[string]string)
 	params["resourceName"] = resourceName
 	params["version"] = query.Get("version")
-	params["consistent"] = query.Get("consistent")
+	if _, ok := query["consistent"]; ok {
+		params["consistent"] = "true"
+	}
 
 	return
 }
diff --git a/internal/resource/http/http_test.go b/internal/resource/http/http_test.go
index c06da46d40bb..97e64e50ca1b 100644
--- a/internal/resource/http/http_test.go
+++ b/internal/resource/http/http_test.go
@@ -255,7 +255,7 @@ func TestResourceWriteHandler(t *testing.T) {
 
 	t.Run("Read resource", func(t *testing.T) {
 		rsp := httptest.NewRecorder()
-		req := httptest.NewRequest("GET", "/demo/v2/artist/keith-urban?partition=default&peer_name=local&namespace=default", nil)
+		req := httptest.NewRequest("GET", "/demo/v2/artist/keith-urban?partition=default&peer_name=local&namespace=default&consistent", nil)
 
 		req.Header.Add("x-consul-token", testACLTokenArtistReadPolicy)
 

From adfab07e9ddf9c3a31f9b2723a323eb33c1ee060 Mon Sep 17 00:00:00 2001
From: Xinyi Wang <xinyi.wang@hashicorp.com>
Date: Thu, 3 Aug 2023 12:45:47 -0700
Subject: [PATCH 21/29] add new test

---
 internal/resource/http/http_test.go | 65 +++++++++++++++++++++++++++++
 1 file changed, 65 insertions(+)

diff --git a/internal/resource/http/http_test.go b/internal/resource/http/http_test.go
index 97e64e50ca1b..265284351c64 100644
--- a/internal/resource/http/http_test.go
+++ b/internal/resource/http/http_test.go
@@ -24,6 +24,7 @@ import (
 
 const testACLTokenArtistReadPolicy = "00000000-0000-0000-0000-000000000001"
 const testACLTokenArtistWritePolicy = "00000000-0000-0000-0000-000000000002"
+const fakeToken = "fake-token"
 
 func parseToken(req *http.Request, token *string) {
 	*token = req.Header.Get("x-Consul-Token")
@@ -268,3 +269,67 @@ func TestResourceWriteHandler(t *testing.T) {
 		require.Equal(t, "Keith Urban", result["data"].(map[string]any)["name"])
 	})
 }
+
+func TestResourceReadHandler(t *testing.T) {
+	aclResolver := &resourceSvc.MockACLResolver{}
+	aclResolver.On("ResolveTokenAndDefaultMeta", testACLTokenArtistReadPolicy, mock.Anything, mock.Anything).
+		Return(svctest.AuthorizerFrom(t, demo.ArtistV1ReadPolicy, demo.ArtistV2ReadPolicy), nil)
+	aclResolver.On("ResolveTokenAndDefaultMeta", testACLTokenArtistWritePolicy, mock.Anything, mock.Anything).
+		Return(svctest.AuthorizerFrom(t, demo.ArtistV1WritePolicy, demo.ArtistV2WritePolicy), nil)
+	aclResolver.On("ResolveTokenAndDefaultMeta", fakeToken, mock.Anything, mock.Anything).
+		Return(svctest.AuthorizerFrom(t, ""), nil)
+
+	client := svctest.RunResourceServiceWithACL(t, aclResolver, demo.RegisterTypes)
+
+	v2ArtistHandler := resourceHandler{
+		resource.Registration{
+			Type:  demo.TypeV2Artist,
+			Proto: &pbdemov2.Artist{},
+		},
+		client,
+		parseToken,
+		hclog.NewNullLogger(),
+	}
+
+	rsp := httptest.NewRecorder()
+	req := httptest.NewRequest("PUT", "/demo/v2/artist/keith-urban?partition=default&peer_name=local&namespace=default", strings.NewReader(`
+		{
+			"metadata": {
+				"foo": "bar"
+			},
+			"data": {
+				"name": "Keith Urban",
+				"genre": "GENRE_COUNTRY"
+			}
+		}
+	`))
+	req.Header.Add("x-consul-token", testACLTokenArtistWritePolicy)
+	v2ArtistHandler.ServeHTTP(rsp, req)
+	require.Equal(t, http.StatusOK, rsp.Result().StatusCode)
+
+	t.Run("Read resource", func(t *testing.T) {
+		rsp := httptest.NewRecorder()
+		req := httptest.NewRequest("GET", "/demo/v2/artist/keith-urban?partition=default&peer_name=local&namespace=default&consistent", nil)
+
+		req.Header.Add("x-consul-token", testACLTokenArtistReadPolicy)
+
+		v2ArtistHandler.ServeHTTP(rsp, req)
+
+		require.Equal(t, http.StatusOK, rsp.Result().StatusCode)
+
+		var result map[string]any
+		require.NoError(t, json.NewDecoder(rsp.Body).Decode(&result))
+		require.Equal(t, "Keith Urban", result["data"].(map[string]any)["name"])
+	})
+
+	t.Run("should be blocked if the token is not authorized", func(t *testing.T) {
+		rsp := httptest.NewRecorder()
+		req := httptest.NewRequest("GET", "/demo/v2/artist/keith-urban?partition=default&peer_name=local&namespace=default&consistent", nil)
+
+		req.Header.Add("x-consul-token", fakeToken)
+
+		v2ArtistHandler.ServeHTTP(rsp, req)
+
+		require.Equal(t, http.StatusForbidden, rsp.Result().StatusCode)
+	})
+}

From 996f9e8435662a59369ced355eeb5ea8793593ac Mon Sep 17 00:00:00 2001
From: Xinyi Wang <xinyi.wang@hashicorp.com>
Date: Thu, 3 Aug 2023 13:46:16 -0700
Subject: [PATCH 22/29] add tests regarding the version query parameter

---
 internal/resource/http/http.go      |  3 ++
 internal/resource/http/http_test.go | 47 +++++++++++++++++++++++++++--
 2 files changed, 47 insertions(+), 3 deletions(-)

diff --git a/internal/resource/http/http.go b/internal/resource/http/http.go
index 45d8d6c2773e..532a91cfcaf8 100644
--- a/internal/resource/http/http.go
+++ b/internal/resource/http/http.go
@@ -155,6 +155,9 @@ func handleResponseError(err error, w http.ResponseWriter, h *resourceHandler) {
 		case codes.PermissionDenied:
 			w.WriteHeader(http.StatusForbidden)
 			h.logger.Info("Failed to write to GRPC resource: User not authenticated", "error", err)
+		case codes.Aborted:
+			w.WriteHeader(http.StatusConflict)
+			h.logger.Info("Failed to write to GRPC resource: the request conflict with the current state of the target resource", "error", err)
 		default:
 			w.WriteHeader(http.StatusInternalServerError)
 			h.logger.Error("Failed to write to GRPC resource", "error", err)
diff --git a/internal/resource/http/http_test.go b/internal/resource/http/http_test.go
index 96da8d1cdb9e..dbc6be31fded 100644
--- a/internal/resource/http/http_test.go
+++ b/internal/resource/http/http_test.go
@@ -66,13 +66,12 @@ func TestResourceHandler_InputValidation(t *testing.T) {
 		},
 		{
 			description: "wrong schema",
-			request: httptest.NewRequest("PUT", "/?partition=default&peer_name=local&namespace=default", strings.NewReader(`
+			request: httptest.NewRequest("PUT", "/keith-urban?partition=default&peer_name=local&namespace=default", strings.NewReader(`
 				{
-					"version": "test_version",
 					"metadata": {
 						"foo": "bar"
 					},
-					"data": {
+					"dada": {
 						"name": "Keith Urban",
 						"genre": "GENRE_COUNTRY"
 					}
@@ -198,6 +197,48 @@ func TestResourceWriteHandler(t *testing.T) {
 		require.Equal(t, "Keith Urban", artist.Name)
 	})
 
+	t.Run("should update the record with version parameter", func(t *testing.T) {
+		rsp := httptest.NewRecorder()
+		req := httptest.NewRequest("PUT", "/demo/v2/artist/keith-urban?partition=default&peer_name=local&namespace=default&version=1", strings.NewReader(`
+			{
+				"metadata": {
+					"foo": "bar"
+				},
+				"data": {
+					"name": "Keith Urban",
+					"genre": "GENRE_COUNTRY"
+				}
+			}
+		`))
+
+		req.Header.Add("x-consul-token", testACLTokenArtistWritePolicy)
+
+		v2ArtistHandler.ServeHTTP(rsp, req)
+
+		require.Equal(t, http.StatusOK, rsp.Result().StatusCode)
+	})
+
+	t.Run("should fail the update if the record's version doesn't match the backend record", func(t *testing.T) {
+		rsp := httptest.NewRecorder()
+		req := httptest.NewRequest("PUT", "/demo/v2/artist/keith-urban?partition=default&peer_name=local&namespace=default&version=1", strings.NewReader(`
+			{
+				"metadata": {
+					"foo": "bar"
+				},
+				"data": {
+					"name": "Keith Urban",
+					"genre": "GENRE_COUNTRY"
+				}
+			}
+		`))
+
+		req.Header.Add("x-consul-token", testACLTokenArtistWritePolicy)
+
+		v2ArtistHandler.ServeHTTP(rsp, req)
+
+		require.Equal(t, http.StatusConflict, rsp.Result().StatusCode)
+	})
+
 	t.Run("should write to the resource backend with owner", func(t *testing.T) {
 		rsp := httptest.NewRecorder()
 		req := httptest.NewRequest("PUT", "/demo/v1/artist/keith-urban-v1?partition=default&peer_name=local&namespace=default", strings.NewReader(`

From 071d612ac82fc33964505a42ef50e3ce33653e1f Mon Sep 17 00:00:00 2001
From: Xinyi Wang <xinyi.wang@hashicorp.com>
Date: Thu, 3 Aug 2023 13:49:23 -0700
Subject: [PATCH 23/29] remove duplicate test case

---
 internal/resource/http/http_test.go | 15 ---------------
 1 file changed, 15 deletions(-)

diff --git a/internal/resource/http/http_test.go b/internal/resource/http/http_test.go
index bf7d7154a909..297f59d4c6f9 100644
--- a/internal/resource/http/http_test.go
+++ b/internal/resource/http/http_test.go
@@ -294,21 +294,6 @@ func TestResourceWriteHandler(t *testing.T) {
 		require.NoError(t, readRsp.Resource.Data.UnmarshalTo(&artist))
 		require.Equal(t, "Keith Urban V1", artist.Name)
 	})
-
-	t.Run("Read resource", func(t *testing.T) {
-		rsp := httptest.NewRecorder()
-		req := httptest.NewRequest("GET", "/demo/v2/artist/keith-urban?partition=default&peer_name=local&namespace=default&consistent", nil)
-
-		req.Header.Add("x-consul-token", testACLTokenArtistReadPolicy)
-
-		v2ArtistHandler.ServeHTTP(rsp, req)
-
-		require.Equal(t, http.StatusOK, rsp.Result().StatusCode)
-
-		var result map[string]any
-		require.NoError(t, json.NewDecoder(rsp.Body).Decode(&result))
-		require.Equal(t, "Keith Urban", result["data"].(map[string]any)["name"])
-	})
 }
 
 func TestResourceReadHandler(t *testing.T) {

From 00bab5144effe177b17cab3115c2edeb7d125466 Mon Sep 17 00:00:00 2001
From: Xinyi Wang <xinyi.wang@hashicorp.com>
Date: Thu, 3 Aug 2023 13:50:55 -0700
Subject: [PATCH 24/29] add test case

---
 internal/resource/http/http_test.go | 12 +++++++++++-
 1 file changed, 11 insertions(+), 1 deletion(-)

diff --git a/internal/resource/http/http_test.go b/internal/resource/http/http_test.go
index 297f59d4c6f9..e766e52c1637 100644
--- a/internal/resource/http/http_test.go
+++ b/internal/resource/http/http_test.go
@@ -275,7 +275,6 @@ func TestResourceWriteHandler(t *testing.T) {
 
 		var result map[string]any
 		require.NoError(t, json.NewDecoder(rsp.Body).Decode(&result))
-
 		require.Equal(t, "Keith Urban V1", result["data"].(map[string]any)["name"])
 		require.Equal(t, "keith-urban-v1", result["id"].(map[string]any)["name"])
 
@@ -348,6 +347,17 @@ func TestResourceReadHandler(t *testing.T) {
 		require.Equal(t, "Keith Urban", result["data"].(map[string]any)["name"])
 	})
 
+	t.Run("should not be found if resource not exist", func(t *testing.T) {
+		rsp := httptest.NewRecorder()
+		req := httptest.NewRequest("GET", "/demo/v2/artist/keith-not-exist?partition=default&peer_name=local&namespace=default&consistent", nil)
+
+		req.Header.Add("x-consul-token", testACLTokenArtistReadPolicy)
+
+		v2ArtistHandler.ServeHTTP(rsp, req)
+
+		require.Equal(t, http.StatusNotFound, rsp.Result().StatusCode)
+	})
+
 	t.Run("should be blocked if the token is not authorized", func(t *testing.T) {
 		rsp := httptest.NewRecorder()
 		req := httptest.NewRequest("GET", "/demo/v2/artist/keith-urban?partition=default&peer_name=local&namespace=default&consistent", nil)

From c043f459abb47f89d7938f7242262cda34a41951 Mon Sep 17 00:00:00 2001
From: Xinyi Wang <xinyi.wang@hashicorp.com>
Date: Thu, 3 Aug 2023 16:40:47 -0700
Subject: [PATCH 25/29] code refactor

---
 .../services/resource/testing/testing.go      | 40 ++-----------------
 internal/resource/http/http_test.go           | 10 +++--
 2 files changed, 10 insertions(+), 40 deletions(-)

diff --git a/agent/grpc-external/services/resource/testing/testing.go b/agent/grpc-external/services/resource/testing/testing.go
index b315f9405faa..e049b229b085 100644
--- a/agent/grpc-external/services/resource/testing/testing.go
+++ b/agent/grpc-external/services/resource/testing/testing.go
@@ -8,6 +8,8 @@ import (
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
 
+	"github.com/hashicorp/go-uuid"
+
 	"github.com/hashicorp/consul/acl"
 	"github.com/hashicorp/consul/acl/resolver"
 	svc "github.com/hashicorp/consul/agent/grpc-external/services/resource"
@@ -17,7 +19,6 @@ import (
 	"github.com/hashicorp/consul/internal/storage/inmem"
 	"github.com/hashicorp/consul/proto-public/pbresource"
 	"github.com/hashicorp/consul/sdk/testutil"
-	"github.com/hashicorp/go-uuid"
 )
 
 func randomACLIdentity(t *testing.T) structs.ACLIdentity {
@@ -47,42 +48,7 @@ func AuthorizerFrom(t *testing.T, policyStrs ...string) resolver.Result {
 // RunResourceService runs a Resource Service for the duration of the test and
 // returns a client to interact with it. ACLs will be disabled.
 func RunResourceService(t *testing.T, registerFns ...func(resource.Registry)) pbresource.ResourceServiceClient {
-	t.Helper()
-
-	backend, err := inmem.NewBackend()
-	require.NoError(t, err)
-
-	ctx, cancel := context.WithCancel(context.Background())
-	t.Cleanup(cancel)
-	go backend.Run(ctx)
-
-	registry := resource.NewRegistry()
-	for _, fn := range registerFns {
-		fn(registry)
-	}
-
-	server := grpc.NewServer()
-
-	svc.NewServer(svc.Config{
-		Backend:     backend,
-		Registry:    registry,
-		Logger:      testutil.Logger(t),
-		ACLResolver: resolver.DANGER_NO_AUTH{},
-	}).Register(server)
-
-	pipe := internal.NewPipeListener()
-	go server.Serve(pipe)
-	t.Cleanup(server.Stop)
-
-	conn, err := grpc.Dial("",
-		grpc.WithTransportCredentials(insecure.NewCredentials()),
-		grpc.WithContextDialer(pipe.DialContext),
-		grpc.WithBlock(),
-	)
-	require.NoError(t, err)
-	t.Cleanup(func() { _ = conn.Close() })
-
-	return pbresource.NewResourceServiceClient(conn)
+	return RunResourceServiceWithACL(t, resolver.DANGER_NO_AUTH{}, registerFns...)
 }
 
 func RunResourceServiceWithACL(t *testing.T, aclResolver svc.ACLResolver, registerFns ...func(resource.Registry)) pbresource.ResourceServiceClient {
diff --git a/internal/resource/http/http_test.go b/internal/resource/http/http_test.go
index dbc6be31fded..d385a861c3d0 100644
--- a/internal/resource/http/http_test.go
+++ b/internal/resource/http/http_test.go
@@ -26,7 +26,7 @@ const testACLTokenArtistReadPolicy = "00000000-0000-0000-0000-000000000001"
 const testACLTokenArtistWritePolicy = "00000000-0000-0000-0000-000000000002"
 
 func parseToken(req *http.Request, token *string) {
-	*token = req.Header.Get("x-Consul-Token")
+	*token = req.Header.Get("X-Consul-Token")
 }
 
 func TestResourceHandler_InputValidation(t *testing.T) {
@@ -205,7 +205,7 @@ func TestResourceWriteHandler(t *testing.T) {
 					"foo": "bar"
 				},
 				"data": {
-					"name": "Keith Urban",
+					"name": "Keith Urban Two",
 					"genre": "GENRE_COUNTRY"
 				}
 			}
@@ -216,9 +216,13 @@ func TestResourceWriteHandler(t *testing.T) {
 		v2ArtistHandler.ServeHTTP(rsp, req)
 
 		require.Equal(t, http.StatusOK, rsp.Result().StatusCode)
+		var result map[string]any
+		require.NoError(t, json.NewDecoder(rsp.Body).Decode(&result))
+		require.Equal(t, "Keith Urban Two", result["data"].(map[string]any)["name"])
+		require.Equal(t, "keith-urban", result["id"].(map[string]any)["name"])
 	})
 
-	t.Run("should fail the update if the record's version doesn't match the backend record", func(t *testing.T) {
+	t.Run("should fail the update if the resource's version doesn't match the version of the existing resource", func(t *testing.T) {
 		rsp := httptest.NewRecorder()
 		req := httptest.NewRequest("PUT", "/demo/v2/artist/keith-urban?partition=default&peer_name=local&namespace=default&version=1", strings.NewReader(`
 			{

From 0a189db8024102f35601f76b3ee8932c72ee965b Mon Sep 17 00:00:00 2001
From: Xinyi Wang <xinyi.wang@hashicorp.com>
Date: Fri, 11 Aug 2023 09:55:17 -0700
Subject: [PATCH 26/29] compare whole body in test

---
 internal/resource/http/http_test.go | 23 ++++++++++++++++++++++-
 1 file changed, 22 insertions(+), 1 deletion(-)

diff --git a/internal/resource/http/http_test.go b/internal/resource/http/http_test.go
index 6ae9b8b4b183..87217560ef81 100644
--- a/internal/resource/http/http_test.go
+++ b/internal/resource/http/http_test.go
@@ -361,7 +361,28 @@ func TestResourceReadHandler(t *testing.T) {
 
 		var result map[string]any
 		require.NoError(t, json.NewDecoder(rsp.Body).Decode(&result))
-		require.Equal(t, "Keith Urban", result["data"].(map[string]any)["name"])
+		// generation and uid are random
+		delete(result, "generation")
+		delete(result["id"].(map[string]interface{}), "uid")
+		expected := map[string]interface{}(map[string]interface{}{
+			"data": map[string]interface{}{"genre": "GENRE_COUNTRY", "name": "Keith Urban"},
+			"id": map[string]interface{}{
+				"name": "keith-urban",
+				"tenancy": map[string]interface{}{
+					"namespace": "default",
+					"partition": "default",
+					"peerName":  "local",
+				},
+				"type": map[string]interface{}{
+					"group":        "demo",
+					"groupVersion": "v2",
+					"kind":         "Artist",
+				},
+			},
+			"metadata": map[string]interface{}{"foo": "bar"},
+			"version":  "1",
+		})
+		require.Equal(t, expected, result)
 	})
 
 	t.Run("should not be found if resource not exist", func(t *testing.T) {

From e04c8df0eb9a59cdf320271bbf553eca49f60bfa Mon Sep 17 00:00:00 2001
From: Xinyi Wang <xinyi.wang@hashicorp.com>
Date: Fri, 11 Aug 2023 10:06:01 -0700
Subject: [PATCH 27/29] refactor

---
 internal/resource/http/http_test.go | 16 +---------------
 1 file changed, 1 insertion(+), 15 deletions(-)

diff --git a/internal/resource/http/http_test.go b/internal/resource/http/http_test.go
index 87217560ef81..89440a349471 100644
--- a/internal/resource/http/http_test.go
+++ b/internal/resource/http/http_test.go
@@ -333,21 +333,7 @@ func TestResourceReadHandler(t *testing.T) {
 		hclog.NewNullLogger(),
 	}
 
-	rsp := httptest.NewRecorder()
-	req := httptest.NewRequest("PUT", "/demo/v2/artist/keith-urban?partition=default&peer_name=local&namespace=default", strings.NewReader(`
-		{
-			"metadata": {
-				"foo": "bar"
-			},
-			"data": {
-				"name": "Keith Urban",
-				"genre": "GENRE_COUNTRY"
-			}
-		}
-	`))
-	req.Header.Add("x-consul-token", testACLTokenArtistWritePolicy)
-	v2ArtistHandler.ServeHTTP(rsp, req)
-	require.Equal(t, http.StatusOK, rsp.Result().StatusCode)
+	createResource(t, v2ArtistHandler)
 
 	t.Run("Read resource", func(t *testing.T) {
 		rsp := httptest.NewRecorder()

From 76a0a9fb887f0c7e110ac74773b3b302e8eace15 Mon Sep 17 00:00:00 2001
From: Xinyi Wang <xinyi.wang@hashicorp.com>
Date: Fri, 11 Aug 2023 10:09:04 -0700
Subject: [PATCH 28/29] linter

---
 internal/resource/http/http_test.go | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/internal/resource/http/http_test.go b/internal/resource/http/http_test.go
index 89440a349471..eb135046cff2 100644
--- a/internal/resource/http/http_test.go
+++ b/internal/resource/http/http_test.go
@@ -350,7 +350,7 @@ func TestResourceReadHandler(t *testing.T) {
 		// generation and uid are random
 		delete(result, "generation")
 		delete(result["id"].(map[string]interface{}), "uid")
-		expected := map[string]interface{}(map[string]interface{}{
+		expected := map[string]interface{}{
 			"data": map[string]interface{}{"genre": "GENRE_COUNTRY", "name": "Keith Urban"},
 			"id": map[string]interface{}{
 				"name": "keith-urban",
@@ -367,7 +367,7 @@ func TestResourceReadHandler(t *testing.T) {
 			},
 			"metadata": map[string]interface{}{"foo": "bar"},
 			"version":  "1",
-		})
+		}
 		require.Equal(t, expected, result)
 	})
 

From 129bd83de9dd2d32ae4675664a6377fbc1ddf35d Mon Sep 17 00:00:00 2001
From: Xinyi Wang <xinyi.wang@hashicorp.com>
Date: Fri, 11 Aug 2023 12:22:59 -0700
Subject: [PATCH 29/29] refactor tests

---
 internal/resource/http/http_test.go | 106 ++++++++--------------------
 1 file changed, 31 insertions(+), 75 deletions(-)

diff --git a/internal/resource/http/http_test.go b/internal/resource/http/http_test.go
index eb135046cff2..07c2b9dbc821 100644
--- a/internal/resource/http/http_test.go
+++ b/internal/resource/http/http_test.go
@@ -110,25 +110,9 @@ func TestResourceWriteHandler(t *testing.T) {
 
 	client := svctest.RunResourceServiceWithACL(t, aclResolver, demo.RegisterTypes)
 
-	v1ArtistHandler := resourceHandler{
-		resource.Registration{
-			Type:  demo.TypeV1Artist,
-			Proto: &pbdemov1.Artist{},
-		},
-		client,
-		parseToken,
-		hclog.NewNullLogger(),
-	}
-
-	v2ArtistHandler := resourceHandler{
-		resource.Registration{
-			Type:  demo.TypeV2Artist,
-			Proto: &pbdemov2.Artist{},
-		},
-		client,
-		parseToken,
-		hclog.NewNullLogger(),
-	}
+	r := resource.NewRegistry()
+	demo.RegisterTypes(r)
+	handler := NewHandler(client, r, parseToken, hclog.NewNullLogger())
 
 	t.Run("should be blocked if the token is not authorized", func(t *testing.T) {
 		rsp := httptest.NewRecorder()
@@ -146,7 +130,7 @@ func TestResourceWriteHandler(t *testing.T) {
 
 		req.Header.Add("x-consul-token", testACLTokenArtistReadPolicy)
 
-		v2ArtistHandler.ServeHTTP(rsp, req)
+		handler.ServeHTTP(rsp, req)
 
 		require.Equal(t, http.StatusForbidden, rsp.Result().StatusCode)
 	})
@@ -167,7 +151,7 @@ func TestResourceWriteHandler(t *testing.T) {
 
 		req.Header.Add("x-consul-token", testACLTokenArtistWritePolicy)
 
-		v2ArtistHandler.ServeHTTP(rsp, req)
+		handler.ServeHTTP(rsp, req)
 
 		require.Equal(t, http.StatusOK, rsp.Result().StatusCode)
 
@@ -207,7 +191,7 @@ func TestResourceWriteHandler(t *testing.T) {
 
 		req.Header.Add("x-consul-token", testACLTokenArtistWritePolicy)
 
-		v2ArtistHandler.ServeHTTP(rsp, req)
+		handler.ServeHTTP(rsp, req)
 
 		require.Equal(t, http.StatusOK, rsp.Result().StatusCode)
 		var result map[string]any
@@ -232,7 +216,7 @@ func TestResourceWriteHandler(t *testing.T) {
 
 		req.Header.Add("x-consul-token", testACLTokenArtistWritePolicy)
 
-		v2ArtistHandler.ServeHTTP(rsp, req)
+		handler.ServeHTTP(rsp, req)
 
 		require.Equal(t, http.StatusConflict, rsp.Result().StatusCode)
 	})
@@ -266,7 +250,7 @@ func TestResourceWriteHandler(t *testing.T) {
 
 		req.Header.Add("x-consul-token", testACLTokenArtistWritePolicy)
 
-		v1ArtistHandler.ServeHTTP(rsp, req)
+		handler.ServeHTTP(rsp, req)
 
 		require.Equal(t, http.StatusOK, rsp.Result().StatusCode)
 
@@ -292,7 +276,7 @@ func TestResourceWriteHandler(t *testing.T) {
 	})
 }
 
-func createResource(t *testing.T, artistHandler resourceHandler) {
+func createResource(t *testing.T, artistHandler http.Handler) map[string]any {
 	rsp := httptest.NewRecorder()
 	req := httptest.NewRequest("PUT", "/demo/v2/artist/keith-urban?partition=default&peer_name=local&namespace=default", strings.NewReader(`
 		{
@@ -310,6 +294,10 @@ func createResource(t *testing.T, artistHandler resourceHandler) {
 
 	artistHandler.ServeHTTP(rsp, req)
 	require.Equal(t, http.StatusOK, rsp.Result().StatusCode)
+
+	var result map[string]any
+	require.NoError(t, json.NewDecoder(rsp.Body).Decode(&result))
+	return result
 }
 
 func TestResourceReadHandler(t *testing.T) {
@@ -323,17 +311,11 @@ func TestResourceReadHandler(t *testing.T) {
 
 	client := svctest.RunResourceServiceWithACL(t, aclResolver, demo.RegisterTypes)
 
-	v2ArtistHandler := resourceHandler{
-		resource.Registration{
-			Type:  demo.TypeV2Artist,
-			Proto: &pbdemov2.Artist{},
-		},
-		client,
-		parseToken,
-		hclog.NewNullLogger(),
-	}
+	r := resource.NewRegistry()
+	demo.RegisterTypes(r)
+	handler := NewHandler(client, r, parseToken, hclog.NewNullLogger())
 
-	createResource(t, v2ArtistHandler)
+	createdResource := createResource(t, handler)
 
 	t.Run("Read resource", func(t *testing.T) {
 		rsp := httptest.NewRecorder()
@@ -341,34 +323,13 @@ func TestResourceReadHandler(t *testing.T) {
 
 		req.Header.Add("x-consul-token", testACLTokenArtistReadPolicy)
 
-		v2ArtistHandler.ServeHTTP(rsp, req)
+		handler.ServeHTTP(rsp, req)
 
 		require.Equal(t, http.StatusOK, rsp.Result().StatusCode)
 
 		var result map[string]any
 		require.NoError(t, json.NewDecoder(rsp.Body).Decode(&result))
-		// generation and uid are random
-		delete(result, "generation")
-		delete(result["id"].(map[string]interface{}), "uid")
-		expected := map[string]interface{}{
-			"data": map[string]interface{}{"genre": "GENRE_COUNTRY", "name": "Keith Urban"},
-			"id": map[string]interface{}{
-				"name": "keith-urban",
-				"tenancy": map[string]interface{}{
-					"namespace": "default",
-					"partition": "default",
-					"peerName":  "local",
-				},
-				"type": map[string]interface{}{
-					"group":        "demo",
-					"groupVersion": "v2",
-					"kind":         "Artist",
-				},
-			},
-			"metadata": map[string]interface{}{"foo": "bar"},
-			"version":  "1",
-		}
-		require.Equal(t, expected, result)
+		require.Equal(t, result, createdResource)
 	})
 
 	t.Run("should not be found if resource not exist", func(t *testing.T) {
@@ -377,7 +338,7 @@ func TestResourceReadHandler(t *testing.T) {
 
 		req.Header.Add("x-consul-token", testACLTokenArtistReadPolicy)
 
-		v2ArtistHandler.ServeHTTP(rsp, req)
+		handler.ServeHTTP(rsp, req)
 
 		require.Equal(t, http.StatusNotFound, rsp.Result().StatusCode)
 	})
@@ -388,7 +349,7 @@ func TestResourceReadHandler(t *testing.T) {
 
 		req.Header.Add("x-consul-token", fakeToken)
 
-		v2ArtistHandler.ServeHTTP(rsp, req)
+		handler.ServeHTTP(rsp, req)
 
 		require.Equal(t, http.StatusForbidden, rsp.Result().StatusCode)
 	})
@@ -403,38 +364,33 @@ func TestResourceDeleteHandler(t *testing.T) {
 
 	client := svctest.RunResourceServiceWithACL(t, aclResolver, demo.RegisterTypes)
 
-	v2ArtistHandler := resourceHandler{
-		resource.Registration{
-			Type:  demo.TypeV2Artist,
-			Proto: &pbdemov2.Artist{},
-		},
-		client,
-		parseToken,
-		hclog.NewNullLogger(),
-	}
+	r := resource.NewRegistry()
+	demo.RegisterTypes(r)
+
+	handler := NewHandler(client, r, parseToken, hclog.NewNullLogger())
 
 	t.Run("should surface PermissionDenied error from resource service", func(t *testing.T) {
-		createResource(t, v2ArtistHandler)
+		createResource(t, handler)
 
 		deleteRsp := httptest.NewRecorder()
 		deletReq := httptest.NewRequest("DELETE", "/demo/v2/artist/keith-urban?partition=default&peer_name=local&namespace=default", strings.NewReader(""))
 
 		deletReq.Header.Add("x-consul-token", testACLTokenArtistReadPolicy)
 
-		v2ArtistHandler.ServeHTTP(deleteRsp, deletReq)
+		handler.ServeHTTP(deleteRsp, deletReq)
 
 		require.Equal(t, http.StatusForbidden, deleteRsp.Result().StatusCode)
 	})
 
 	t.Run("should delete a resource without version", func(t *testing.T) {
-		createResource(t, v2ArtistHandler)
+		createResource(t, handler)
 
 		deleteRsp := httptest.NewRecorder()
 		deletReq := httptest.NewRequest("DELETE", "/demo/v2/artist/keith-urban?partition=default&peer_name=local&namespace=default", strings.NewReader(""))
 
 		deletReq.Header.Add("x-consul-token", testACLTokenArtistWritePolicy)
 
-		v2ArtistHandler.ServeHTTP(deleteRsp, deletReq)
+		handler.ServeHTTP(deleteRsp, deletReq)
 
 		require.Equal(t, http.StatusNoContent, deleteRsp.Result().StatusCode)
 
@@ -453,14 +409,14 @@ func TestResourceDeleteHandler(t *testing.T) {
 	})
 
 	t.Run("should delete a resource with version", func(t *testing.T) {
-		createResource(t, v2ArtistHandler)
+		createResource(t, handler)
 
 		rsp := httptest.NewRecorder()
 		req := httptest.NewRequest("DELETE", "/demo/v2/artist/keith-urban?partition=default&peer_name=local&namespace=default&version=1", strings.NewReader(""))
 
 		req.Header.Add("x-consul-token", testACLTokenArtistWritePolicy)
 
-		v2ArtistHandler.ServeHTTP(rsp, req)
+		handler.ServeHTTP(rsp, req)
 
 		require.Equal(t, http.StatusNoContent, rsp.Result().StatusCode)