feat: added support for optimistic locking with version number (#19)

**Description**

Added support for optimistic locking with version number for improved
application concurrency behavior when using `PutItem` operation to
create/update items.

> Optimistic locking is a strategy to ensure that the client-side item
that you are updating (or deleting) is the same as the item in Amazon
DynamoDB. If you use this strategy, your database writes are protected
from being overwritten by the writes of others, and vice versa.

https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/DynamoDBMapper.OptimisticLocking.html

**How it works**

With optimistic locking, each item has an attribute that acts as a
version number. If you retrieve an item from a table, the application
records the version number of that item. You can update the item, but
only if the version number on the server side has not changed. If there
is a version mismatch, it means that someone else has modified the item
before you did. The update attempt fails, because you have a stale
version of the item. If this happens, try again by retrieving the item
and then trying to update it. Optimistic locking prevents you from
accidentally overwriting changes that were made by others. It also
prevents others from accidentally overwriting your changes.

_Clients needs to implement the retry behavior on their end for this to
work. Simply retrying on this library to increment version number
defeats the purpose of ensuring data integrity_

**Usage**

Provide `WithOptimisticLock` option when calling PutItem method.

```go
type LedgerAccount struct {
	ID      string
	Balance int
	Version uint
}

func AddBalance(ctx context.Context, acc LedgerAccount, amount int) (err error) {
	var try int
	for try <= maxTries {
		var acc LedgerAccount
		err, _ := table.GetItem(ctx, pk, pk, &acc)
		if err != nil {
			return err
		}

		// Add amount to current account balance
		acc.Balance += amount

		// If another go routine updates account using AddBalance we want to avoid overwriting using an old balance
		err = table.PutItem(ctx, pk, pk, acc, dynago.WithOptimisticLock("Version", acc.Version))
		if err == nil {
			return nil
		}
		// Retry if there is an error with latest item value from DynamoDB
		try += 1
	}

	return err
}
```

Author: shidil

README.md

@@ -89,6 +89,50 @@ err := table.PutItem(
 )
 ```
 
+#### Optimistic locking with version number
+
+> Optimistic locking is a strategy to ensure that the client-side item that you are updating (or deleting) is the same as the item in Amazon DynamoDB.
+If you use this strategy, your database writes are protected from being overwritten by the writes of others, and vice versa.
+
+Use the `WithOptimisticLock` option when calling `PutItem` method.
+
+This works well and is recommended when using the event sourcing pattern where you need to update aggregate snapshots.
+You can make use of event broker retry mechanism or retry libraries to simplify retry
+
+**Example**
+
+```go
+type LedgerAccount struct {
+	ID      string
+	Balance int
+	Version uint
+}
+
+func AddBalance(ctx context.Context, acc LedgerAccount, amount int) (err error) {
+	var try int
+	for try <= maxTries {
+		var acc LedgerAccount
+		err, _ := table.GetItem(ctx, pk, pk, &acc)
+		if err != nil {
+			return err
+		}
+
+		// Add amount to current account balance
+		acc.Balance += amount
+
+		// If another go routine updates account using AddBalance we want to avoid overwriting using an old balance
+		err = table.PutItem(ctx, pk, pk, acc, dynago.WithOptimisticLock("Version", acc.Version))
+		if err == nil {
+			return nil
+		}
+		// Retry if there is an error with latest item value from DynamoDB
+		try += 1
+	}
+
+	return err
+}
+```
+
 ### Query
 
 ```go

interface.go

@@ -24,8 +24,8 @@ func BoolValue(v bool) *types.AttributeValueMemberBOOL {
 type WriteAPI interface {
 	// Create or update given item in DynamoDB. Must implemenmt DynamoRecord interface.
 	// DynamoRecord.GetKeys will be called to get values for parition and sort keys.
-	PutItem(ctx context.Context, pk Attribute, sk Attribute, item interface{}) error
-	DeleteItem(ctx context.Context, pk string, sk string) error
+	PutItem(ctx context.Context, pk, sk Attribute, item interface{}, opt ...PutOption) error
+	DeleteItem(ctx context.Context, pk, sk string) error
 	BatchDeleteItems(ctx context.Context, input []AttributeRecord) []AttributeRecord
 }
 
@@ -35,7 +35,7 @@ type TransactionAPI interface {
 }
 
 type ReadAPI interface {
-	GetItem(ctx context.Context, pk Attribute, sk Attribute, out interface{}) (error, bool)
+	GetItem(ctx context.Context, pk, sk Attribute, out interface{}) (error, bool)
 	BatchGetItems(ctx context.Context, input []AttributeRecord, out interface{}) error
 }
 

put_item.go

@@ -10,12 +10,37 @@ import (
 	"github.com/aws/aws-sdk-go-v2/service/dynamodb/types"
 )
 
+type PutOption func(*dynamodb.PutItemInput) error
+
+// Enables concurrency control by using an optimistic lock
+// https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/DynamoDBMapper.OptimisticLocking.html
+//
+// Provide key field acts as a version number (Usually called Version)
+// GetItem retrieves current version number and you can update the item if the version number in DynamoDB hasn't changed
+// Each update increments the version number and if the update fails fetch the record again to get latest version number and try again
+func WithOptimisticLock(key string, currentVersion uint) PutOption {
+	return func(input *dynamodb.PutItemInput) error {
+		condition := "#version = :oldVersion"
+		input.ConditionExpression = &condition
+		if input.ExpressionAttributeNames == nil {
+			input.ExpressionAttributeNames = map[string]string{}
+		}
+		if input.ExpressionAttributeValues == nil {
+			input.ExpressionAttributeValues = map[string]Attribute{}
+		}
+		input.ExpressionAttributeNames["#version"] = key
+		input.ExpressionAttributeValues[":oldVersion"] = NumberValue(int64(currentVersion))
+		input.Item[key] = NumberValue(int64(currentVersion + 1))
+		return nil
+	}
+}
+
 /**
 * Used to put and update a db record from dynamodb given a partition key and sort key
 * @param item the item put into the database
  * @return true if the record was put, false otherwise
 */
-func (t *Client) PutItem(ctx context.Context, pk Attribute, sk Attribute, item interface{}) error {
+func (t *Client) PutItem(ctx context.Context, pk, sk Attribute, item interface{}, opts ...PutOption) error {
 	av, err := attributevalue.MarshalMap(item)
 	if err != nil {
 		log.Println("Failed to Marshal item" + err.Error())
@@ -26,10 +51,18 @@ func (t *Client) PutItem(ctx context.Context, pk Attribute, sk Attribute, item i
 		av[k] = v
 	}
 
-	_, err = t.client.PutItem(ctx, &dynamodb.PutItemInput{
+	input := &dynamodb.PutItemInput{
 		TableName: &t.TableName,
 		Item:      av,
-	})
+	}
+	// Apply option functions
+	if len(opts) > 0 {
+		for _, opt := range opts {
+			opt(input)
+		}
+	}
+
+	_, err = t.client.PutItem(ctx, input)
 	if err != nil {
 		log.Println("Failed to Put item" + err.Error())
 		return err

tests/putitem_test.go

@@ -0,0 +1,162 @@
+package tests
+
+import (
+	"context"
+	"fmt"
+	"reflect"
+	"strings"
+	"sync"
+	"testing"
+
+	"github.com/oolio-group/dynago"
+)
+
+type Record struct {
+	ID string
+	Pk string
+	Sk string
+}
+
+func TestPutItem(t *testing.T) {
+	table := prepareTable(t, dynamoEndpoint, "put_test")
+	testCases := []struct {
+		title       string
+		item        Record
+		expected    Record
+		expectedErr error
+	}{
+		{
+			title: "success 1",
+			item: Record{
+				ID: "1",
+				Pk: "account_1",
+				Sk: "account_1",
+			},
+			expected: Record{
+				ID: "1",
+				Pk: "account_1",
+				Sk: "account_1",
+			},
+		},
+		{
+			title: "success 2",
+			item: Record{
+				ID: "2",
+				Pk: "account_2",
+				Sk: "account_2",
+			},
+			expected: Record{
+				ID: "2",
+				Pk: "account_2",
+				Sk: "account_2",
+			},
+		},
+		{
+			title: "pk is required",
+			item: Record{
+				ID: "3",
+				Sk: "account_3",
+			},
+			expectedErr: fmt.Errorf("The AttributeValue for a key attribute cannot contain an empty string value. Key: pk"),
+		},
+		{
+			title: "sk is required",
+			item: Record{
+				ID: "4",
+				Pk: "account_4",
+			},
+			expectedErr: fmt.Errorf("The AttributeValue for a key attribute cannot contain an empty string value. Key: sk"),
+		},
+	}
+	for _, tc := range testCases {
+		t.Run(tc.title, func(t *testing.T) {
+			t.Helper()
+			ctx := context.TODO()
+
+			pk := dynago.StringValue(tc.item.Pk)
+			sk := dynago.StringValue(tc.item.Sk)
+			err := table.PutItem(ctx, pk, sk, &tc.item)
+			if err != nil {
+				if tc.expectedErr == nil {
+					t.Fatalf("unexpected error %s", err)
+				}
+				if !strings.Contains(err.Error(), tc.expectedErr.Error()) {
+					t.Fatalf("expected op to fail with %s; got %s", tc.expectedErr, err)
+				}
+				return
+			}
+
+			var out Record
+			err, found := table.GetItem(ctx, pk, sk, &out)
+			if err != nil {
+				t.Fatalf("unexpected error %s", err)
+			}
+			if !found {
+				t.Errorf("expected to find item with pk %s and sk %s", tc.item.Pk, tc.item.Sk)
+			}
+			if !reflect.DeepEqual(tc.expected, out) {
+				t.Errorf("expected query to return %v; got %v", tc.expected, out)
+			}
+		})
+	}
+}
+
+type LedgerAccount struct {
+	ID      string
+	Balance int
+	Version uint
+}
+
+func TestPutItemWithOptimisticLock(t *testing.T) {
+	table := prepareTable(t, dynamoEndpoint, "put_optimistic_test")
+	// Create new account item in DynamoDB with default values
+	account := LedgerAccount{ID: "123"}
+	ctx := context.Background()
+	pk := dynago.StringValue("123")
+	err := table.PutItem(ctx, pk, pk, account)
+	if err != nil {
+		t.Fatalf("unexpected error %s", err)
+		return
+	}
+
+	// Update method will add 100 to the current account balance
+	update := func() error {
+		var (
+			acc LedgerAccount
+		)
+		err, _ := table.GetItem(ctx, pk, pk, &acc)
+		if err != nil {
+			return err
+		}
+		t.Log(acc)
+		acc.Balance += 100
+		return table.PutItem(ctx, pk, pk, acc, dynago.WithOptimisticLock("Version", acc.Version))
+	}
+	// Invoke Update in parallel 10 times to increment account balance
+	// WithOptimisticLock should prevent concurrency overwriting issues
+	var wg sync.WaitGroup
+	for range 10 {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for {
+				err := update()
+				if err == nil {
+					return
+				}
+			}
+		}()
+	}
+	wg.Wait()
+	// We expect account balance to be 1000 after 10 update
+	// If any update method overwrote with an outdated value then total balance will be less than 1000
+	var acc LedgerAccount
+	err, _ = table.GetItem(ctx, pk, pk, &acc)
+	if err != nil {
+		t.Fatalf("unexpected error %s", err)
+		return
+	}
+	if acc.Balance != 1000 {
+		t.Errorf("expected account balance to be 1000 after 10 increments of 100; got %d", acc.Balance)
+	}
+}