Rewrite statemachine (#35)

* Rewrite statemachine

To support all states, the statemachine needed a refactor. This ended up
being more of a full rewrite using the lessons learned from the first
version of the state machine.

* Removed old statemachine and rewire http handlers.

Not e2e tested yet.

* Finish up statemachine rewrite

- Create small reconciliation loop
- Add logging
- Add colourful logging when in debug mode.
- Fix bugs

* Fix linter errors

.mockery.yaml

@@ -0,0 +1,19 @@
+# Copyright 2024 go-dataspace
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+with-expecter: true
+packages:
+  github.com/go-dataspace/run-dsrpc/gen/go/provider/v1:
+    interfaces:
+      ProviderServiceClient:

Makefile

@@ -1,11 +1,11 @@
 # Copyright 2024 go-dataspace
-# 
+#
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
 # You may obtain a copy of the License at
-# 
+#
 #     https://www.apache.org/licenses/LICENSE-2.0
-# 
+#
 # Unless required by applicable law or agreed to in writing, software
 # distributed under the License is distributed on an "AS IS" BASIS,
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
@@ -33,3 +33,6 @@ lint:
 
 vulncheck:
 	govulncheck ./...
+
+generate:
+	go generate ./...

dsp/catalog_handlers.go

@@ -16,7 +16,6 @@ package dsp
 
 import (
 	"fmt"
-	"io"
 	"net/http"
 	"time"
 
@@ -38,14 +37,9 @@ var dataService = shared.DataService{
 
 func (ch *dspHandlers) catalogRequestHandler(w http.ResponseWriter, req *http.Request) {
 	logger := logging.Extract(req.Context())
-	body, err := io.ReadAll(req.Body)
+	catalogReq, err := shared.DecodeValid[shared.CatalogRequestMessage](req)
 	if err != nil {
-		returnError(w, http.StatusBadRequest, "Could not read body")
-		return
-	}
-	catalogReq, err := shared.UnmarshalAndValidate(req.Context(), body, shared.CatalogRequestMessage{})
-	if err != nil {
-		logger.Error("Non validating catalog request", "error", err)
+		logger.Error("Non validating catalog request", "err", err)
 		returnError(w, http.StatusBadRequest, "Request did not validate")
 		return
 	}
@@ -57,7 +51,7 @@ func (ch *dspHandlers) catalogRequestHandler(w http.ResponseWriter, req *http.Re
 		return
 	}
 
-	validateMarshalAndReturn(req.Context(), w, http.StatusOK, shared.CatalogAcknowledgement{
+	err = shared.EncodeValid(w, req, http.StatusOK, shared.CatalogAcknowledgement{
 		Dataset: shared.Dataset{
 			Resource: shared.Resource{
 				ID:   "urn:uuid:3afeadd8-ed2d-569e-d634-8394a8836d57",
@@ -72,6 +66,9 @@ func (ch *dspHandlers) catalogRequestHandler(w http.ResponseWriter, req *http.Re
 		Datasets: processProviderCatalogue(resp.GetDatasets(), dataService),
 		Service:  []shared.DataService{dataService},
 	})
+	if err != nil {
+		logger.Error("failed to serve catalog", "err", err)
+	}
 }
 
 func (ch *dspHandlers) datasetRequestHandler(w http.ResponseWriter, req *http.Request) {
@@ -83,18 +80,13 @@ func (ch *dspHandlers) datasetRequestHandler(w http.ResponseWriter, req *http.Re
 	ctx, logger := logging.InjectLabels(req.Context(), "paramID", paramID)
 	id, err := uuid.Parse(paramID)
 	if err != nil {
-		logger.Error("Misformed uuid in path", "error", err)
+		logger.Error("Misformed uuid in path", "err", err)
 		returnError(w, http.StatusBadRequest, "Invalid ID")
 		return
 	}
-	body, err := io.ReadAll(req.Body)
+	datasetReq, err := shared.DecodeValid[shared.DatasetRequestMessage](req)
 	if err != nil {
-		returnError(w, http.StatusBadRequest, "Could not read body")
-		return
-	}
-	datasetReq, err := shared.UnmarshalAndValidate(ctx, body, shared.DatasetRequestMessage{})
-	if err != nil {
-		logger.Error("Non validating dataset request", "error", err)
+		logger.Error("Non validating dataset request", "err", err)
 		returnError(w, http.StatusBadRequest, "Request did not validate")
 		return
 	}
@@ -107,7 +99,10 @@ func (ch *dspHandlers) datasetRequestHandler(w http.ResponseWriter, req *http.Re
 		return
 	}
 
-	validateMarshalAndReturn(req.Context(), w, http.StatusOK, processProviderDataset(resp.GetDataset(), dataService))
+	err = shared.EncodeValid(w, req, http.StatusOK, processProviderDataset(resp.GetDataset(), dataService))
+	if err != nil {
+		logger.Error("failed to serve dataset", "err", err)
+	}
 }
 
 func processProviderDataset(pds *providerv1.Dataset, service shared.DataService) shared.Dataset {

dsp/common_handlers.go

@@ -15,13 +15,14 @@
 package dsp
 
 import (
-	"context"
 	"encoding/json"
 	"fmt"
 	"log/slog"
 	"net/http"
+	"net/url"
 
 	"github.com/go-dataspace/run-dsp/dsp/shared"
+	"github.com/go-dataspace/run-dsp/dsp/statemachine"
 	"github.com/go-dataspace/run-dsp/internal/constants"
 	"github.com/go-dataspace/run-dsp/jsonld"
 	providerv1 "github.com/go-dataspace/run-dsrpc/gen/go/provider/v1"
@@ -30,7 +31,10 @@ import (
 )
 
 type dspHandlers struct {
-	provider providerv1.ProviderServiceClient
+	store      statemachine.Archiver
+	provider   providerv1.ProviderServiceClient
+	reconciler *statemachine.Reconciler
+	selfURL    *url.URL
 }
 
 type errorResponse struct {
@@ -51,15 +55,6 @@ func returnContent(w http.ResponseWriter, status int, content string) {
 	fmt.Fprint(w, content)
 }
 
-func validateMarshalAndReturn[T any](ctx context.Context, w http.ResponseWriter, successStatus int, s T) {
-	respBody, err := shared.ValidateAndMarshal(ctx, s)
-	if err != nil {
-		returnError(w, http.StatusInternalServerError, "Could not render response")
-		return
-	}
-	returnContent(w, successStatus, string(respBody))
-}
-
 func returnError(w http.ResponseWriter, status int, e string) {
 	errResp := errorString(e)
 	returnContent(w, status, errResp)
@@ -72,7 +67,7 @@ func routeNotImplemented(w http.ResponseWriter, req *http.Request) {
 }
 
 func grpcErrorHandler(w http.ResponseWriter, l *slog.Logger, err error) {
-	l.Error("Got GRPC error", "error", err)
+	l.Error("Got GRPC error", "err", err)
 	switch status.Code(err) { //nolint:exhaustive
 	case codes.Unauthenticated:
 		returnContent(w, http.StatusForbidden, "not authenticated")
@@ -83,7 +78,7 @@ func grpcErrorHandler(w http.ResponseWriter, l *slog.Logger, err error) {
 	case codes.NotFound:
 		returnContent(w, http.StatusNotFound, "not found")
 	default:
-		returnContent(w, http.StatusInternalServerError, "error")
+		returnContent(w, http.StatusInternalServerError, "err")
 	}
 }