Dynamodb datastore

go.mod

@@ -29,9 +29,13 @@ require (
 	github.com/authzed/authzed-go v1.4.1
 	github.com/authzed/consistent v0.1.0
 	github.com/authzed/grpcutil v0.0.0-20240123194739-2ea1e3d2d98b
-	github.com/aws/aws-sdk-go-v2 v1.36.4
+	github.com/aws/aws-sdk-go v1.55.8
+	github.com/aws/aws-sdk-go-v2 v1.37.1
 	github.com/aws/aws-sdk-go-v2/config v1.29.16
+	github.com/aws/aws-sdk-go-v2/feature/dynamodb/attributevalue v1.20.1
+	github.com/aws/aws-sdk-go-v2/feature/dynamodb/expression v1.8.1
 	github.com/aws/aws-sdk-go-v2/feature/rds/auth v1.5.12
+	github.com/aws/aws-sdk-go-v2/service/dynamodb v1.45.1
 	github.com/benbjohnson/clock v1.3.5
 	github.com/bits-and-blooms/bloom/v3 v3.7.0
 	github.com/caio/go-tdigest/v4 v4.0.1
@@ -118,7 +122,11 @@ require (
 	sigs.k8s.io/controller-runtime v0.21.0
 )
 
-require golang.org/x/vuln v1.1.4 // indirect
+require (
+	github.com/aws/aws-sdk-go-v2/service/dynamodbstreams v1.27.1 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/endpoint-discovery v1.11.1 // indirect
+	golang.org/x/vuln v1.1.4 // indirect
+)
 
 // Most tools are managed in the magefiles module. These tools are just
 // the ones that can't run from a submodule at the moment.
@@ -176,15 +184,15 @@ require (
 	github.com/authzed/ctxkey v0.0.0-20250226155515-d49f99185584
 	github.com/aws/aws-sdk-go-v2/credentials v1.17.69 // indirect
 	github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.16.31 // indirect
-	github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.35 // indirect
-	github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.35 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.1 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.1 // indirect
 	github.com/aws/aws-sdk-go-v2/internal/ini v1.8.3 // indirect
-	github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.12.3 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.0 // indirect
 	github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.12.16 // indirect
 	github.com/aws/aws-sdk-go-v2/service/sso v1.25.4 // indirect
 	github.com/aws/aws-sdk-go-v2/service/ssooidc v1.30.2 // indirect
 	github.com/aws/aws-sdk-go-v2/service/sts v1.33.21 // indirect
-	github.com/aws/smithy-go v1.22.2 // indirect
+	github.com/aws/smithy-go v1.22.5 // indirect
 	github.com/aymanbagabas/go-osc52/v2 v2.0.1 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/bits-and-blooms/bitset v1.10.0 // indirect

go.sum

@@ -1455,24 +1455,36 @@ github.com/authzed/ctxkey v0.0.0-20250226155515-d49f99185584 h1:mP7EpcUL90EKcf/F
 github.com/authzed/ctxkey v0.0.0-20250226155515-d49f99185584/go.mod h1:wnimjr5RPPouIhZQ3ztDBLMUKKuUroj3U9Jy0PxeaEE=
 github.com/authzed/grpcutil v0.0.0-20240123194739-2ea1e3d2d98b h1:wbh8IK+aMLTCey9sZasO7b6BWLAJnHHvb79fvWCXwxw=
 github.com/authzed/grpcutil v0.0.0-20240123194739-2ea1e3d2d98b/go.mod h1:s3qC7V7XIbiNWERv7Lfljy/Lx25/V1Qlexb0WJuA8uQ=
-github.com/aws/aws-sdk-go-v2 v1.36.4 h1:GySzjhVvx0ERP6eyfAbAuAXLtAda5TEy19E5q5W8I9E=
-github.com/aws/aws-sdk-go-v2 v1.36.4/go.mod h1:LLXuLpgzEbD766Z5ECcRmi8AzSwfZItDtmABVkRLGzg=
+github.com/aws/aws-sdk-go v1.55.8 h1:JRmEUbU52aJQZ2AjX4q4Wu7t4uZjOu71uyNmaWlUkJQ=
+github.com/aws/aws-sdk-go v1.55.8/go.mod h1:ZkViS9AqA6otK+JBBNH2++sx1sgxrPKcSzPPvQkUtXk=
+github.com/aws/aws-sdk-go-v2 v1.37.1 h1:SMUxeNz3Z6nqGsXv0JuJXc8w5YMtrQMuIBmDx//bBDY=
+github.com/aws/aws-sdk-go-v2 v1.37.1/go.mod h1:9Q0OoGQoboYIAJyslFyF1f5K1Ryddop8gqMhWx/n4Wg=
 github.com/aws/aws-sdk-go-v2/config v1.29.16 h1:XkruGnXX1nEZ+Nyo9v84TzsX+nj86icbFAeust6uo8A=
 github.com/aws/aws-sdk-go-v2/config v1.29.16/go.mod h1:uCW7PNjGwZ5cOGZ5jr8vCWrYkGIhPoTNV23Q/tpHKzg=
 github.com/aws/aws-sdk-go-v2/credentials v1.17.69 h1:8B8ZQboRc3uaIKjshve/XlvJ570R7BKNy3gftSbS178=
 github.com/aws/aws-sdk-go-v2/credentials v1.17.69/go.mod h1:gPME6I8grR1jCqBFEGthULiolzf/Sexq/Wy42ibKK9c=
+github.com/aws/aws-sdk-go-v2/feature/dynamodb/attributevalue v1.20.1 h1:1ToPL5M0nYwkIOTb9r+ION0ZZe9xemRe1mRMWMw5ihs=
+github.com/aws/aws-sdk-go-v2/feature/dynamodb/attributevalue v1.20.1/go.mod h1:dDdNpGWZdj4AxADkfM1IG1IutBmSJM7zURhUNOVv/lE=
+github.com/aws/aws-sdk-go-v2/feature/dynamodb/expression v1.8.1 h1:W5pbxu4HaA56dHdCdBgDk0UccIMtYVZIZvQTXNlgOd8=
+github.com/aws/aws-sdk-go-v2/feature/dynamodb/expression v1.8.1/go.mod h1:nbLpEe6EynGLqFEkvq1K2TFqxozTYC2sqIUWg0gS/ZI=
 github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.16.31 h1:oQWSGexYasNpYp4epLGZxxjsDo8BMBh6iNWkTXQvkwk=
 github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.16.31/go.mod h1:nc332eGUU+djP3vrMI6blS0woaCfHTe3KiSQUVTMRq0=
 github.com/aws/aws-sdk-go-v2/feature/rds/auth v1.5.12 h1:TGacjTPZ5GNFTltGPNH76EX9g7ZX6prlyBxc9SqHuKw=
 github.com/aws/aws-sdk-go-v2/feature/rds/auth v1.5.12/go.mod h1:t10ExEjdhPVHJNyzQh12+xnW8G1wb5bwnoqZ+657kp8=
-github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.35 h1:o1v1VFfPcDVlK3ll1L5xHsaQAFdNtZ5GXnNR7SwueC4=
-github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.35/go.mod h1:rZUQNYMNG+8uZxz9FOerQJ+FceCiodXvixpeRtdESrU=
-github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.35 h1:R5b82ubO2NntENm3SAm0ADME+H630HomNJdgv+yZ3xw=
-github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.35/go.mod h1:FuA+nmgMRfkzVKYDNEqQadvEMxtxl9+RLT9ribCwEMs=
+github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.1 h1:ksZXBYv80EFTcgc8OJO48aQ8XDWXIQL7gGasPeCoTzI=
+github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.1/go.mod h1:HSksQyyJETVZS7uM54cir0IgxttTD+8aEoJMPGepHBI=
+github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.1 h1:+dn/xF/05utS7tUhjIcndbuaPjfll2LhbH1cCDGLYUQ=
+github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.1/go.mod h1:hyAGz30LHdm5KBZDI58MXx5lDVZ5CUfvfTZvMu4HCZo=
 github.com/aws/aws-sdk-go-v2/internal/ini v1.8.3 h1:bIqFDwgGXXN1Kpp99pDOdKMTTb5d2KyU5X/BZxjOkRo=
 github.com/aws/aws-sdk-go-v2/internal/ini v1.8.3/go.mod h1:H5O/EsxDWyU+LP/V8i5sm8cxoZgc2fdNR9bxlOFrQTo=
-github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.12.3 h1:eAh2A4b5IzM/lum78bZ590jy36+d/aFLgKF/4Vd1xPE=
-github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.12.3/go.mod h1:0yKJC/kb8sAnmlYa6Zs3QVYqaC8ug2AbnNChv5Ox3uA=
+github.com/aws/aws-sdk-go-v2/service/dynamodb v1.45.1 h1:gFD9BLrXox2Q5zxFwyD2OnGb40YYofQ/anaGxVP848Q=
+github.com/aws/aws-sdk-go-v2/service/dynamodb v1.45.1/go.mod h1:J+qJkxNypYjDcwXldBH+ox2T7OshtP6LOq5VhU0v6hg=
+github.com/aws/aws-sdk-go-v2/service/dynamodbstreams v1.27.1 h1:H4W48E0/zjiHLlL59/Y0DpaB+krXsuarjwrquCwMtT4=
+github.com/aws/aws-sdk-go-v2/service/dynamodbstreams v1.27.1/go.mod h1:nGsqtVMMjTeFot6U+rLj+mpOcZybPoxyQPMKY4GHwQo=
+github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.0 h1:6+lZi2JeGKtCraAj1rpoZfKqnQ9SptseRZioejfUOLM=
+github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.0/go.mod h1:eb3gfbVIxIoGgJsi9pGne19dhCBpK6opTYpQqAmdy44=
+github.com/aws/aws-sdk-go-v2/service/internal/endpoint-discovery v1.11.1 h1:/E4JUPMI8LRX2XpXsbmKN42l1lZPoLjGJ/Kun97pLc0=
+github.com/aws/aws-sdk-go-v2/service/internal/endpoint-discovery v1.11.1/go.mod h1:qgbd/t8S8y5e87KPQ4kC0kyxZ0K6nC1QiDtFMoxlsOo=
 github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.12.16 h1:/ldKrPPXTC421bTNWrUIpq3CxwHwRI/kpc+jPUTJocM=
 github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.12.16/go.mod h1:5vkf/Ws0/wgIMJDQbjI4p2op86hNW6Hie5QtebrDgT8=
 github.com/aws/aws-sdk-go-v2/service/sso v1.25.4 h1:EU58LP8ozQDVroOEyAfcq0cGc5R/FTZjVoYJ6tvby3w=
@@ -1481,8 +1493,8 @@ github.com/aws/aws-sdk-go-v2/service/ssooidc v1.30.2 h1:XB4z0hbQtpmBnb1FQYvKaCM7
 github.com/aws/aws-sdk-go-v2/service/ssooidc v1.30.2/go.mod h1:hwRpqkRxnQ58J9blRDrB4IanlXCpcKmsC83EhG77upg=
 github.com/aws/aws-sdk-go-v2/service/sts v1.33.21 h1:nyLjs8sYJShFYj6aiyjCBI3EcLn1udWrQTjEF+SOXB0=
 github.com/aws/aws-sdk-go-v2/service/sts v1.33.21/go.mod h1:EhdxtZ+g84MSGrSrHzZiUm9PYiZkrADNja15wtRJSJo=
-github.com/aws/smithy-go v1.22.2 h1:6D9hW43xKFrRx/tXXfAlIZc4JI+yQe6snnWcQyxSyLQ=
-github.com/aws/smithy-go v1.22.2/go.mod h1:irrKGvNn1InZwb2d7fkIRNucdfwR8R+Ts3wxYa/cJHg=
+github.com/aws/smithy-go v1.22.5 h1:P9ATCXPMb2mPjYBgueqJNCA5S9UfktsW0tTxi+a7eqw=
+github.com/aws/smithy-go v1.22.5/go.mod h1:t1ufH5HMublsJYulve2RKmHDC15xu1f26kHCp/HgceI=
 github.com/aymanbagabas/go-osc52/v2 v2.0.1 h1:HwpRHbFMcZLEVr42D4p7XBqjyuxQH5SMiErDT4WkJ2k=
 github.com/aymanbagabas/go-osc52/v2 v2.0.1/go.mod h1:uYgXzlJ7ZpABp8OJ+exZzJJhRNQ2ASbcXHWsFqH8hp8=
 github.com/benbjohnson/clock v1.1.0/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA=
@@ -2011,6 +2023,8 @@ github.com/jingyugao/rowserrcheck v1.1.1 h1:zibz55j/MJtLsjP1OF4bSdgXxwL1b+Vn7Tjz
 github.com/jingyugao/rowserrcheck v1.1.1/go.mod h1:4yvlZSDb3IyDTUZJUmpZfm2Hwok+Dtp+nu2qOq+er9c=
 github.com/jjti/go-spancheck v0.6.5 h1:lmi7pKxa37oKYIMScialXUK6hP3iY5F1gu+mLBPgYB8=
 github.com/jjti/go-spancheck v0.6.5/go.mod h1:aEogkeatBrbYsyW6y5TgDfihCulDYciL1B7rG2vSsrU=
+github.com/jmespath/go-jmespath v0.4.0 h1:BEgLn5cpjn8UN1mAw4NjwDrS35OdebyEtFe+9YPoQUg=
+github.com/jmespath/go-jmespath v0.4.0/go.mod h1:T8mJZnbsbmF+m6zOOFylbeCJqk5+pHWvzYPziyZiYoo=
 github.com/joho/godotenv v1.5.1 h1:7eLL/+HRGLY0ldzfGMeQkb7vMd0as4CfYvUVzLqw0N0=
 github.com/joho/godotenv v1.5.1/go.mod h1:f4LDr5Voq0i2e/R5DDNOoa2zzDfwtkZa6DnEwAbqwq4=
 github.com/josharian/intern v1.0.0 h1:vlS4z54oSdjm0bgjRigI+G1HpF+tI+9rE5LLzOg8HmY=

internal/datastore/dynamodb/caveat.go

@@ -0,0 +1,249 @@
+package dynamodb
+
+import (
+	"context"
+	"fmt"
+
+	sq "github.com/Masterminds/squirrel"
+	"github.com/authzed/spicedb/pkg/datastore"
+	corev1 "github.com/authzed/spicedb/pkg/proto/core/v1"
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/feature/dynamodb/expression"
+	"github.com/aws/aws-sdk-go-v2/service/dynamodb"
+	ddbv2 "github.com/aws/aws-sdk-go-v2/service/dynamodb"
+	"github.com/aws/aws-sdk-go-v2/service/dynamodb/types"
+)
+
+// WriteCaveats implements datastore.ReadWriteTransaction.
+func (d DynamodbReadWriterTx) WriteCaveats(ctx context.Context, caveats []*corev1.CaveatDefinition) error {
+
+	wr := []types.WriteRequest{}
+	for _, caveat := range caveats {
+		serialized, err := caveat.MarshalVT()
+		if err != nil {
+			return fmt.Errorf("%d", err)
+		}
+
+		kvp := KeyValues{}
+
+		// fmt.Printf("%#t\n", caveat)
+
+		kvp[ColCaveat] = &caveat.Name
+		kvp[ColCreatedXid] = aws.String(d.xid.String())
+		kvp[ColDeletedXid] = aws.String("")
+
+		wr = append(wr, types.WriteRequest{
+			PutRequest: Caveat.GetPutItem(kvp, map[string]interface{}{
+				ColSerialized: serialized,
+			}),
+		})
+
+	}
+	d.ds.client.BatchWriteItem(ctx, &dynamodb.BatchWriteItemInput{
+		RequestItems: map[string][]types.WriteRequest{
+			TableName: wr,
+		},
+	})
+
+	return nil
+}
+
+// ReadCaveatByName implements datastore.ReadWriteTransaction.
+func (d dynamodbReader) ReadCaveatByName(ctx context.Context, name string) (caveat *corev1.CaveatDefinition, lastWritten datastore.Revision, err error) {
+
+	key := expression.KeyEqual(expression.Key(SK), expression.Value(Caveat.SK.Build(
+		KeyValues{
+			ColCaveat: &name,
+		})))
+
+	key = key.And(expression.KeyEqual(expression.Key(PK), expression.Value(Caveat.PK.Build(
+		KeyValues{
+			ColEntity: &Caveat.Entity,
+		}))))
+
+	expr, err := expression.NewBuilder().
+		WithKeyCondition(key).
+		Build()
+	if err != nil {
+		return nil, nil, err
+	}
+
+	res, err := d.ds.QueryItem(ctx, &expr)
+
+	item, ok := res[0].(map[string]interface{})
+	if !ok {
+		return nil, nil, fmt.Errorf("")
+	}
+
+	serializedAttr, exists := item[ColSerialized]
+	if !exists {
+		return nil, nil, fmt.Errorf("serialized attribute not found")
+	}
+
+	binaryAttr, ok := serializedAttr.([]byte)
+	if !ok {
+		return nil, nil, fmt.Errorf("serialized attribute is not binary type")
+	}
+
+	caveat = &corev1.CaveatDefinition{}
+
+	err = caveat.UnmarshalVT(binaryAttr)
+	if err != nil {
+		return nil, nil, fmt.Errorf("failed to unmarshal protobuf: %w", err)
+	}
+
+	return caveat, nil, nil
+}
+
+// LookupCaveatsWithNames implements datastore.ReadWriteTransaction.
+func (d dynamodbReader) LookupCaveatsWithNames(ctx context.Context, names []string) ([]datastore.RevisionedCaveat, error) {
+	if len(names) == 0 {
+		return nil, fmt.Errorf("names list cannot be empty")
+	}
+
+	values := []expression.OperandBuilder{}
+	for _, name := range names {
+		n := Caveat.LSI1SK.Build(KeyValues{
+			ColCaveat: &name,
+		})
+		values = append(values, expression.Value(n))
+	}
+	var filter expression.ConditionBuilder
+
+	filter = expression.Name(LSI1SK).In(values[0], values[1:]...)
+
+	key := expression.KeyEqual(expression.Key(PK), expression.Value(Caveat.PK.Build(KeyValues{
+		ColEntity: &Caveat.Entity,
+	})))
+
+	expr, err := expression.NewBuilder().
+		WithKeyCondition(key).
+		WithFilter(filter).
+		Build()
+	if err != nil {
+		return nil, err
+	}
+
+	res, err := d.ds.QueryItem(ctx, &expr)
+
+	resNs := []datastore.RevisionedCaveat{}
+
+	for _, r := range res {
+		item, ok := r.(map[string]interface{})
+		if !ok {
+			return nil, fmt.Errorf("")
+		}
+
+		serializedAttr, exists := item[ColSerialized]
+		if !exists {
+			return nil, fmt.Errorf("serialized attribute not found")
+		}
+
+		binaryAttr, ok := serializedAttr.([]byte)
+		if !ok {
+			return nil, fmt.Errorf("serialized attribute is not binary type")
+		}
+
+		caveat := &corev1.CaveatDefinition{}
+
+		err := caveat.UnmarshalVT(binaryAttr)
+		if err != nil {
+			return nil, fmt.Errorf("failed to unmarshal protobuf: %w", err)
+		}
+
+		resNs = append(resNs, datastore.RevisionedCaveat{
+			Definition:          caveat,
+			LastWrittenRevision: datastore.NoRevision,
+		})
+	}
+
+	return resNs, nil
+}
+
+func (d dynamodbReader) ListAllCaveats(ctx context.Context) ([]datastore.RevisionedCaveat, error) {
+	key := expression.KeyEqual(expression.Key(PK), expression.Value(Caveat.PK.Build(KeyValues{
+		ColEntity: &Caveat.Entity,
+	})))
+
+	expr, err := expression.NewBuilder().
+		WithKeyCondition(key).
+		Build()
+	if err != nil {
+		return nil, err
+	}
+
+	res, err := d.ds.QueryItem(ctx, &expr)
+
+	resNs := []datastore.RevisionedCaveat{}
+
+	for _, r := range res {
+		item, ok := r.(map[string]interface{})
+		if !ok {
+			return nil, fmt.Errorf("")
+		}
+
+		serializedAttr, exists := item[ColSerialized]
+		if !exists {
+			return nil, fmt.Errorf("serialized attribute not found")
+		}
+
+		binaryAttr, ok := serializedAttr.([]byte)
+		if !ok {
+			return nil, fmt.Errorf("serialized attribute is not binary type")
+		}
+
+		caveat := &corev1.CaveatDefinition{}
+
+		err := caveat.UnmarshalVT(binaryAttr)
+		if err != nil {
+			return nil, fmt.Errorf("failed to unmarshal protobuf: %w", err)
+		}
+
+		resNs = append(resNs, datastore.RevisionedCaveat{
+			Definition:          caveat,
+			LastWrittenRevision: datastore.NoRevision,
+		})
+	}
+
+	return resNs, nil
+}
+
+// DeleteCaveats implements datastore.ReadWriteTransaction.
+func (d DynamodbReadWriterTx) DeleteCaveats(ctx context.Context, names []string) error {
+	statements := []types.BatchStatementRequest{}
+
+	for _, name := range names {
+		deleteCaveat := delete
+		query := deleteCaveat.
+			Where(sq.Eq{PK: Caveat.PK.Build(KeyValues{ColEntity: &Caveat.Entity})}).
+			Where(sq.Eq{SK: Caveat.SK.Build(KeyValues{ColCaveat: &name})})
+
+		sql, args, err := query.ToSql()
+		if err != nil {
+			return err
+		}
+		parameters := []types.AttributeValue{}
+		for _, v := range args {
+			parameters = append(parameters, &types.AttributeValueMemberS{
+				Value: v.(string),
+			})
+		}
+
+		statements = append(statements, types.BatchStatementRequest{
+			Statement:  aws.String(sql),
+			Parameters: parameters,
+		})
+	}
+
+	res, err := d.ds.client.BatchExecuteStatement(ctx, &ddbv2.BatchExecuteStatementInput{
+		Statements: statements,
+	})
+	if err != nil {
+		fmt.Println(err.Error())
+		return err
+	}
+
+	fmt.Printf("output - %#v\n", res)
+
+	return nil
+}