feat: middleware

.github/workflows/tests.yml

@@ -17,6 +17,10 @@ jobs:
     - uses: actions/setup-go@v4
       with:
         go-version-file: 'go.mod'
+    - uses: brokeyourbike/[email protected]
+      with:
+        mockery-version: '2.36.0'
+    - run: mockery --all --inpackage --quiet
     - run: go build -v ./...
     - run: go test -race -covermode=atomic -coverprofile=coverage.out -v ./...
 

.gitignore

@@ -14,4 +14,5 @@
 # Dependency directories (remove the comment below to include it)
 # vendor/
 
-.DS_Store
+.DS_Store
+mock_*.go

go.mod

@@ -60,6 +60,7 @@ require (
 	github.com/spf13/afero v1.10.0 // indirect
 	github.com/spf13/cast v1.5.1 // indirect
 	github.com/spf13/pflag v1.0.5 // indirect
+	github.com/stretchr/objx v0.5.1 // indirect
 	github.com/subosito/gotenv v1.6.0 // indirect
 	github.com/twitchyliquid64/golang-asm v0.15.1 // indirect
 	github.com/ugorji/go/codec v1.2.11 // indirect

go.sum

@@ -265,6 +265,8 @@ github.com/spf13/viper v1.17.0/go.mod h1:BmMMMLQXSbcHK6KAOiFLz0l5JHrU89OdIRHvsk0
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
 github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo=
+github.com/stretchr/objx v0.5.1 h1:4VhoImhV/Bm0ToFkXFi8hXNXwpDRZ/ynw3amt82mzq0=
+github.com/stretchr/objx v0.5.1/go.mod h1:/iHQpkQwBD6DLUmQ4pE+s1TXdob1mORJ4/UFdrifcy0=
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
 github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
 github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA=

middleware/pubsub.go

@@ -0,0 +1,58 @@
+package middleware
+
+import (
+	"context"
+	"net/http"
+
+	"github.com/gin-gonic/gin"
+	"github.com/glocurrency/commons/logger"
+	"github.com/glocurrency/commons/q"
+)
+
+const PubSubMessageCtx = "pubSubMessageCtx"
+
+type Locker interface {
+	TryToLock(ctx context.Context, key string) error
+}
+
+type PubSubCtx struct {
+	locker Locker
+}
+
+func NewPubSubCtx(l Locker) *PubSubCtx {
+	return &PubSubCtx{locker: l}
+}
+
+func (m *PubSubCtx) Middleware() gin.HandlerFunc {
+	return func(ctx *gin.Context) {
+		var msg q.PubSubMessage
+		if err := ctx.ShouldBindJSON(&msg); err != nil {
+			logger.WithContext(ctx).WithError(err).Error("cannot unmarshal pubsub message")
+			ctx.AbortWithStatus(http.StatusBadRequest)
+			return
+		}
+
+		logger.WithContext(ctx).WithField("pubsub_message", msg).Debug("pubsub message received")
+
+		if msg.GetUniqueKey() != "" {
+			if err := m.locker.TryToLock(ctx, msg.GetUniqueKey()); err != nil {
+				logger.WithContext(ctx).
+					WithError(err).
+					WithField("unique_key", msg.GetUniqueKey()).
+					Error("cannot lock task")
+
+				ctx.AbortWithStatus(http.StatusUnprocessableEntity)
+				return
+			}
+		}
+
+		ctx.Set(PubSubMessageCtx, msg)
+		ctx.Next()
+	}
+}
+
+// MustGetMessageFromContext returns the PubSub message from the context.
+func MustGetMessageFromContext(ctx *gin.Context) q.PubSubMessage {
+	msg := ctx.MustGet(PubSubMessageCtx).(q.PubSubMessage)
+	return msg
+}

middleware/pubsubs_test.go

@@ -0,0 +1,96 @@
+package middleware_test
+
+import (
+	"bytes"
+	_ "embed"
+	"errors"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/gin-gonic/gin"
+	"github.com/glocurrency/commons/middleware"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+)
+
+//go:embed testdata/pubsub-valid.json
+var validPubSubMsg []byte
+
+//go:embed testdata/pubsub-valid-unique.json
+var validPubSubUniqueMsg []byte
+
+//go:embed testdata/pubsub-invalid.json
+var invalidPubSubMsg []byte
+
+func TestPubSubCtx(t *testing.T) {
+	tests := []struct {
+		name       string
+		body       []byte
+		wantStatus int
+		setupMock  func(lockerMock *middleware.MockLocker)
+	}{
+		{
+			"success without lock",
+			validPubSubMsg,
+			http.StatusOK,
+			nil,
+		},
+		{
+			"success with lock",
+			validPubSubUniqueMsg,
+			http.StatusOK,
+			func(lockerMock *middleware.MockLocker) {
+				lockerMock.On("TryToLock", mock.Anything, "unique-456").Return(nil)
+			},
+		},
+		{
+			"invalid message",
+			invalidPubSubMsg,
+			http.StatusBadRequest,
+			nil,
+		},
+		{
+			"cannot lock by unique the key",
+			validPubSubUniqueMsg,
+			http.StatusUnprocessableEntity,
+			func(lockerMock *middleware.MockLocker) {
+				lockerMock.On("TryToLock", mock.Anything, "unique-456").Return(errors.New("I am an error"))
+			},
+		},
+	}
+
+	for i := range tests {
+		test := tests[i]
+		t.Run(test.name, func(t *testing.T) {
+			t.Parallel()
+
+			tasksMock := middleware.NewMockLocker(t)
+			if test.setupMock != nil {
+				test.setupMock(tasksMock)
+			}
+
+			m := middleware.NewPubSubCtx(tasksMock)
+			req := httptest.NewRequest(http.MethodPost, "/", nil)
+
+			if test.body != nil {
+				req.Body = io.NopCloser(bytes.NewReader(test.body))
+			}
+
+			w := httptest.NewRecorder()
+			router := gin.New()
+			router.Use(m.Middleware())
+			router.POST("/", func(ctx *gin.Context) {
+				msg := middleware.MustGetMessageFromContext(ctx)
+				assert.NotNil(t, msg)
+
+				ctx.String(http.StatusOK, "the end.")
+			})
+			router.ServeHTTP(w, req)
+
+			require.Equal(t, test.wantStatus, w.Code)
+		})
+	}
+}

middleware/testdata/pubsub-invalid.json

@@ -0,0 +1,3 @@
+{
+    "message": "I am not a valid pubsub payload"
+}

middleware/testdata/pubsub-valid-unique.json

@@ -0,0 +1,10 @@
+{
+    "message": {
+        "messageId": "158b5c5d-ce6b-4020-b822-ceb0e43b754a",
+        "publishTime": "2023-07-09T17:49:48.087822341Z",
+        "attributes": {
+            "uniqueKey": "unique-456"
+        }
+    },
+    "subscription": "sub-123"
+}

middleware/testdata/pubsub-valid.json

@@ -0,0 +1,7 @@
+{
+    "message": {
+        "messageId": "529ebc2c-99c6-47b0-9bf6-a92c17c8d35a",
+        "publishTime": "2023-07-09T17:47:54.896272657Z"
+    },
+    "subscription": "sub-123"
+}