refactor: enhance code coverage (#28)

Earthfile

@@ -89,7 +89,9 @@ protogen:
     # generate the pbs
     RUN buf generate \
             --template buf.gen.yaml \
-            --path protos/internal
+            --path protos/internal \
+            --path protos/test
 
     # save artifact to
     SAVE ARTIFACT gen/internal AS LOCAL internal/internalpb
+    SAVE ARTIFACT gen/test AS LOCAL  test/data/testpb

README.md

@@ -26,11 +26,9 @@ go get github.com/tochemey/gokv
     - `Put`: create key/value pair that is eventually distributed in the cluster of nodes. The `key` is a string and the `value` is a byte array.
     - `PutProto`: to create a key/value pair where the value is a protocol buffer message
     - `PutString`: to create a key/value pair where the value is a string
-    - `PutAny`: to create a key/value pair with a given [`Codec`](./cluster/codec.go) to encode the value type.
     - `Get`: retrieves the value of a given `key` from the cluster of nodes.
     - `GetProto`: retrieves a protocol buffer message for a given `key`. This requires `PutProto` or `Put` to be used to set the value.
     - `GetString`: retrieves a string value for a given `key`. This requires `PutString` or `Put` to be used to set the value.
-    - `GetAny`: retrieves any value type for a given `key`. This requires `PutAny` to be used to set the value.
     - `Exists`: check the existence of a given `key` in the cluster.
     - `Delete`: delete a given `key` from the cluster. At the moment the `key` is marked to be `archived`.
 

cluster/client.go

@@ -76,16 +76,6 @@ func (client *Client) PutString(ctx context.Context, key string, value string, e
 	return client.Put(ctx, key, []byte(value), expiration)
 }
 
-// PutAny distributes the key/value pair in the cluster.
-// A binary encoder is required to properly encode the value.
-func (client *Client) PutAny(ctx context.Context, key string, value any, expiration time.Duration, codec Codec) error {
-	bytea, err := codec.Encode(value)
-	if err != nil {
-		return err
-	}
-	return client.Put(ctx, key, bytea, expiration)
-}
-
 // Get retrieves the value of the given key from the cluster
 func (client *Client) Get(ctx context.Context, key string) ([]byte, error) {
 	if !client.connected.Load() {
@@ -99,7 +89,7 @@ func (client *Client) Get(ctx context.Context, key string) ([]byte, error) {
 	if err != nil {
 		code := connect.CodeOf(err)
 		if code == connect.CodeNotFound {
-			return nil, nil
+			return nil, ErrKeyNotFound
 		}
 		return nil, err
 	}
@@ -119,23 +109,13 @@ func (client *Client) GetProto(ctx context.Context, key string, dst proto.Messag
 
 // GetString retrieves the value of the given from the cluster as a string
 // Prior to calling this method one must set a string as the value of the key
-func (client *Client) GetString(ctx context.Context, key string, dst string) error {
+func (client *Client) GetString(ctx context.Context, key string) (string, error) {
 	bytea, err := client.Get(ctx, key)
 	if err != nil {
-		return err
+		return "", err
 	}
-	dst = string(bytea)
-	return nil
-}
 
-// GetAny retrieves the value of the given from the cluster
-// Prior to calling this method one must set a string as the value of the key
-func (client *Client) GetAny(ctx context.Context, key string, codec Codec) (any, error) {
-	bytea, err := client.Get(ctx, key)
-	if err != nil {
-		return nil, err
-	}
-	return codec.Decode(bytea)
+	return string(bytea), nil
 }
 
 // Delete deletes a given key from the cluster

cluster/client_test.go

@@ -0,0 +1,213 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2024 Tochemey
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package cluster
+
+import (
+	"context"
+	"encoding/json"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/protobuf/proto"
+
+	"github.com/tochemey/gokv/internal/lib"
+	"github.com/tochemey/gokv/test/data/testpb"
+)
+
+func TestClient(t *testing.T) {
+	t.Run("With PutProto GetProto", func(t *testing.T) {
+		ctx := context.Background()
+		// start the NATS server
+		srv := startNatsServer(t)
+		// create a cluster node1
+		node1, sd1 := startNode(t, srv.Addr().String())
+		require.NotNil(t, node1)
+
+		// create a cluster node2
+		node2, sd2 := startNode(t, srv.Addr().String())
+		require.NotNil(t, node2)
+
+		key := "my-key"
+		value := new(testpb.Hello)
+		err := node2.Client().PutProto(ctx, key, value, NoExpiration)
+		require.NoError(t, err)
+
+		// wait for the key to be distributed in the cluster
+		lib.Pause(time.Second)
+
+		// let us retrieve the key from the other nodes
+		exists, err := node1.Client().Exists(ctx, key)
+		require.NoError(t, err)
+		require.True(t, exists)
+
+		actual := &testpb.Hello{}
+		err = node1.Client().GetProto(ctx, key, actual)
+		require.NoError(t, err)
+
+		assert.True(t, proto.Equal(value, actual))
+
+		lib.Pause(time.Second)
+		t.Cleanup(func() {
+			assert.NoError(t, node1.Stop(ctx))
+			assert.NoError(t, node2.Stop(ctx))
+			assert.NoError(t, sd1.Close())
+			assert.NoError(t, sd2.Close())
+			srv.Shutdown()
+		})
+	})
+	t.Run("With PutString GetString", func(t *testing.T) {
+		ctx := context.Background()
+		// start the NATS server
+		srv := startNatsServer(t)
+		// create a cluster node1
+		node1, sd1 := startNode(t, srv.Addr().String())
+		require.NotNil(t, node1)
+
+		// create a cluster node2
+		node2, sd2 := startNode(t, srv.Addr().String())
+		require.NotNil(t, node2)
+
+		key := "my-key"
+		value := "my-value"
+		err := node2.Client().PutString(ctx, key, value, NoExpiration)
+		require.NoError(t, err)
+
+		// wait for the key to be distributed in the cluster
+		lib.Pause(time.Second)
+
+		// let us retrieve the key from the other nodes
+		exists, err := node1.Client().Exists(ctx, key)
+		require.NoError(t, err)
+		require.True(t, exists)
+
+		actual, err := node1.Client().GetString(ctx, key)
+		require.NoError(t, err)
+		require.NotEmpty(t, actual)
+		require.Equal(t, value, actual)
+
+		lib.Pause(time.Second)
+		t.Cleanup(func() {
+			assert.NoError(t, node1.Stop(ctx))
+			assert.NoError(t, node2.Stop(ctx))
+			assert.NoError(t, sd1.Close())
+			assert.NoError(t, sd2.Close())
+			srv.Shutdown()
+		})
+	})
+	t.Run("With PutProto GetProto with expiration", func(t *testing.T) {
+		ctx := context.Background()
+		// start the NATS server
+		srv := startNatsServer(t)
+		// create a cluster node1
+		node1, sd1 := startNode(t, srv.Addr().String())
+		require.NotNil(t, node1)
+
+		// create a cluster node2
+		node2, sd2 := startNode(t, srv.Addr().String())
+		require.NotNil(t, node2)
+
+		expiration := 100 * time.Millisecond
+		key := "my-key"
+		value := &testpb.Hello{Name: key}
+		err := node2.Client().PutProto(ctx, key, value, expiration)
+		require.NoError(t, err)
+
+		// wait for the key to be distributed in the cluster
+		lib.Pause(time.Second)
+
+		// let us retrieve the key from the other nodes
+		exists, err := node1.Client().Exists(ctx, key)
+		require.NoError(t, err)
+		require.False(t, exists)
+
+		actual := &testpb.Hello{}
+		err = node1.Client().GetProto(ctx, key, actual)
+		require.Error(t, err)
+		assert.EqualError(t, err, ErrKeyNotFound.Error())
+
+		lib.Pause(time.Second)
+		t.Cleanup(func() {
+			assert.NoError(t, node1.Stop(ctx))
+			assert.NoError(t, node2.Stop(ctx))
+			assert.NoError(t, sd1.Close())
+			assert.NoError(t, sd2.Close())
+			srv.Shutdown()
+		})
+	})
+	t.Run("With PutString GetString with expiration", func(t *testing.T) {
+		ctx := context.Background()
+		// start the NATS server
+		srv := startNatsServer(t)
+		// create a cluster node1
+		node1, sd1 := startNode(t, srv.Addr().String())
+		require.NotNil(t, node1)
+
+		// create a cluster node2
+		node2, sd2 := startNode(t, srv.Addr().String())
+		require.NotNil(t, node2)
+
+		key := "my-key"
+		value := "my-value"
+		expiration := 100 * time.Millisecond
+		err := node2.Client().PutString(ctx, key, value, expiration)
+		require.NoError(t, err)
+
+		// wait for the key to be distributed in the cluster
+		lib.Pause(time.Second)
+
+		// let us retrieve the key from the other nodes
+		exists, err := node1.Client().Exists(ctx, key)
+		require.NoError(t, err)
+		require.False(t, exists)
+
+		actual, err := node1.Client().GetString(ctx, key)
+		require.Error(t, err)
+		require.Empty(t, actual)
+		assert.EqualError(t, err, ErrKeyNotFound.Error())
+
+		lib.Pause(time.Second)
+		t.Cleanup(func() {
+			assert.NoError(t, node1.Stop(ctx))
+			assert.NoError(t, node2.Stop(ctx))
+			assert.NoError(t, sd1.Close())
+			assert.NoError(t, sd2.Close())
+			srv.Shutdown()
+		})
+	})
+}
+
+type testCodec struct{}
+
+func (c *testCodec) Encode(t interface{}) ([]byte, error) {
+	return json.Marshal(t)
+}
+
+func (c *testCodec) Decode(bytea []byte) (interface{}, error) {
+	var t interface{}
+	err := json.Unmarshal(bytea, &t)
+	return t, err
+}

cluster/delegate.go

@@ -231,21 +231,21 @@ func (d *Delegate) Put(key string, value []byte, expiration time.Duration) {
 }
 
 // Get returns the value of the given key
-func (d *Delegate) Get(key string) []byte {
+func (d *Delegate) Get(key string) ([]byte, error) {
 	d.RLock()
 	defer d.RUnlock()
 	localState := d.fsm
 	for _, nodeState := range localState.GetNodeStates() {
 		for k, entry := range nodeState.GetEntries() {
 			if k == key {
 				if expired(entry) {
-					return nil
+					return nil, ErrKeyNotFound
 				}
-				return entry.GetValue()
+				return entry.GetValue(), nil
 			}
 		}
 	}
-	return nil
+	return nil, ErrKeyNotFound
 }
 
 // Delete deletes the given key from the cluster
@@ -277,7 +277,7 @@ func (d *Delegate) Exists(key string) bool {
 				if expired(entry) {
 					return false
 				}
-				return !entry.GetArchived() && len(entry.GetValue()) > 0
+				return !entry.GetArchived()
 			}
 		}
 	}

cluster/node.go

@@ -207,10 +207,10 @@ func (node *Node) Get(ctx context.Context, request *connect.Request[internalpb.G
 	}
 
 	req := request.Msg
-	entry := node.delegate.Get(req.GetKey())
-	if len(entry) == 0 {
+	entry, err := node.delegate.Get(req.GetKey())
+	if err != nil {
 		node.mu.Unlock()
-		return nil, connect.NewError(connect.CodeNotFound, ErrKeyNotFound)
+		return nil, connect.NewError(connect.CodeNotFound, err)
 	}
 
 	node.mu.Unlock()

go.mod

@@ -6,6 +6,7 @@ require (
 	connectrpc.com/connect v1.16.2
 	github.com/deckarep/golang-set/v2 v2.6.0
 	github.com/flowchartsman/retry v1.2.0
+	github.com/golang/protobuf v1.5.4
 	github.com/hashicorp/memberlist v0.5.1
 	github.com/nats-io/nats-server/v2 v2.10.20
 	github.com/nats-io/nats.go v1.37.0
@@ -32,7 +33,6 @@ require (
 	github.com/go-openapi/jsonreference v0.20.2 // indirect
 	github.com/go-openapi/swag v0.22.4 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
-	github.com/golang/protobuf v1.5.4 // indirect
 	github.com/google/btree v1.1.3 // indirect
 	github.com/google/gnostic-models v0.6.8 // indirect
 	github.com/google/go-cmp v0.6.0 // indirect

protos/test/test.proto

@@ -0,0 +1,9 @@
+syntax = "proto3";
+
+package testpb;
+
+option go_package = "github.com/tochemey/gokv/test/data;testpb";
+
+message Hello {
+  string name = 1;
+}