mirror of
https://github.com/tiennm99/tiennm99bot.git
synced 2026-10-15 02:17:35 +00:00
feat(storage): add DynamoDB KV provider sibling to Firestore
- Implement DynamoDB client wrapper and KV provider - Full CRUD test coverage for both local and AWS modes - AWS SDK v2 dependency added to go.mod/go.sum
This commit is contained in:
tiennm99 committed
1 parent
5bcb643222
commit
b061ab78df
7 files changed
+552
-1
No files matched your search
@@ -4,8 +4,13 @@ go 1.25.0
|
||||
|
||||
require (
|
||||
cloud.google.com/go/firestore v1.22.0
|
||||
github.com/aws/aws-sdk-go-v2 v1.41.7
|
||||
github.com/aws/aws-sdk-go-v2/config v1.32.17
|
||||
github.com/aws/aws-sdk-go-v2/service/dynamodb v1.57.3
|
||||
github.com/go-telegram/bot v1.20.0
|
||||
golang.org/x/time v0.15.0
|
||||
google.golang.org/api v0.274.0
|
||||
google.golang.org/genai v1.56.0
|
||||
google.golang.org/grpc v1.80.0
|
||||
)
|
||||
|
||||
@@ -15,13 +20,28 @@ require (
|
||||
cloud.google.com/go/auth/oauth2adapt v0.2.8 // indirect
|
||||
cloud.google.com/go/compute/metadata v0.9.0 // indirect
|
||||
cloud.google.com/go/longrunning v0.9.0 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/credentials v1.19.16 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.23 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.23 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.23 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.24 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.9 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/service/internal/endpoint-discovery v1.11.23 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.23 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/service/signin v1.0.11 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/service/sso v1.30.17 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/service/ssooidc v1.35.21 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/service/sts v1.42.1 // indirect
|
||||
github.com/aws/smithy-go v1.25.1 // indirect
|
||||
github.com/cespare/xxhash/v2 v2.3.0 // indirect
|
||||
github.com/felixge/httpsnoop v1.0.4 // indirect
|
||||
github.com/go-logr/logr v1.4.3 // indirect
|
||||
github.com/go-logr/stdr v1.2.2 // indirect
|
||||
github.com/google/go-cmp v0.7.0 // indirect
|
||||
github.com/google/s2a-go v0.1.9 // indirect
|
||||
github.com/googleapis/enterprise-certificate-proxy v0.3.14 // indirect
|
||||
github.com/googleapis/gax-go/v2 v2.21.0 // indirect
|
||||
github.com/gorilla/websocket v1.5.3 // indirect
|
||||
go.opentelemetry.io/auto/sdk v1.2.1 // indirect
|
||||
go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.61.0 // indirect
|
||||
go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.61.0 // indirect
|
||||
@@ -34,7 +54,6 @@ require (
|
||||
golang.org/x/sync v0.20.0 // indirect
|
||||
golang.org/x/sys v0.44.0 // indirect
|
||||
golang.org/x/text v0.37.0 // indirect
|
||||
golang.org/x/time v0.15.0 // indirect
|
||||
google.golang.org/genproto v0.0.0-20260319201613-d00831a3d3e7 // indirect
|
||||
google.golang.org/genproto/googleapis/api v0.0.0-20260401024825-9d38bb4040a9 // indirect
|
||||
google.golang.org/genproto/googleapis/rpc v0.0.0-20260401024825-9d38bb4040a9 // indirect
|
||||
|
||||
@@ -10,6 +10,38 @@ cloud.google.com/go/firestore v1.22.0 h1:avooeboIq37vKXobrbPUFhFBxS/c3FqmWoX0xs8
|
||||
cloud.google.com/go/firestore v1.22.0/go.mod h1:PaM4i7i7ruALSKmlpHXXZaPObcZw0W7ie5UOPr72iTU=
|
||||
cloud.google.com/go/longrunning v0.9.0 h1:0EzbDEGsAvOZNbqXopgniY0w0a1phvu5IdUFq8grmqY=
|
||||
cloud.google.com/go/longrunning v0.9.0/go.mod h1:pkTz846W7bF4o2SzdWJ40Hu0Re+UoNT6Q5t+igIcb8E=
|
||||
github.com/aws/aws-sdk-go-v2 v1.41.7 h1:DWpAJt66FmnnaRIOT/8ASTucrvuDPZASqhhLey6tLY8=
|
||||
github.com/aws/aws-sdk-go-v2 v1.41.7/go.mod h1:4LAfZOPHNVNQEckOACQx60Y8pSRjIkNZQz1w92xpMJc=
|
||||
github.com/aws/aws-sdk-go-v2/config v1.32.17 h1:FpL4/758/diKwqbytU0prpuiu60fgXKUWCpDJtApclU=
|
||||
github.com/aws/aws-sdk-go-v2/config v1.32.17/go.mod h1:OXqUMzgXytfoF9JaKkhrOYsyh72t9G+MJH8mMRaexOE=
|
||||
github.com/aws/aws-sdk-go-v2/credentials v1.19.16 h1:r3RJBuU7X9ibt8RHbMjWE6y60QbKBiII6wSrXnapxSU=
|
||||
github.com/aws/aws-sdk-go-v2/credentials v1.19.16/go.mod h1:6cx7zqDENJDbBIIWX6P8s0h6hqHC8Avbjh9Dseo27ug=
|
||||
github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.23 h1:UuSfcORqNSz/ey3VPRS8TcVH2Ikf0/sC+Hdj400QI6U=
|
||||
github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.23/go.mod h1:+G/OSGiOFnSOkYloKj/9M35s74LgVAdJBSD5lsFfqKg=
|
||||
github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.23 h1:GpT/TrnBYuE5gan2cZbTtvP+JlHsutdmlV2YfEyNde0=
|
||||
github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.23/go.mod h1:xYWD6BS9ywC5bS3sz9Xh04whO/hzK2plt2Zkyrp4JuA=
|
||||
github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.23 h1:bpd8vxhlQi2r1hiueOw02f/duEPTMK59Q4QMAoTTtTo=
|
||||
github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.23/go.mod h1:15DfR2nw+CRHIk0tqNyifu3G1YdAOy68RftkhMDDwYk=
|
||||
github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.24 h1:OQqn11BtaYv1WLUowvcA30MpzIu8Ti4pcLPIIyoKZrA=
|
||||
github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.24/go.mod h1:X5ZJyfwVrWA96GzPmUCWFQaEARPR7gCrpq2E92PJwAE=
|
||||
github.com/aws/aws-sdk-go-v2/service/dynamodb v1.57.3 h1:XgjzLEE8CrNYnr4Xmi1W5PfKsKMjp4Pu1rWkJNO43JI=
|
||||
github.com/aws/aws-sdk-go-v2/service/dynamodb v1.57.3/go.mod h1:r7sfLXEN8RUA89tAHy1E7lCtVOOWIkqVy/FbnUdxW1E=
|
||||
github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.9 h1:FLudkZLt5ci0ozzgkVo8BJGwvqNaZbTWb3UcucAateA=
|
||||
github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.9/go.mod h1:w7wZ/s9qK7c8g4al+UyoF1Sp/Z45UwMGcqIzLWVQHWk=
|
||||
github.com/aws/aws-sdk-go-v2/service/internal/endpoint-discovery v1.11.23 h1:3Eo/PBBnjFi1+gYfaL286dpmFSW3mTfodBIybq36Qv4=
|
||||
github.com/aws/aws-sdk-go-v2/service/internal/endpoint-discovery v1.11.23/go.mod h1:3oh+5xGSd1iuxonVb3Qbm+WJYlbhczT9kbzr6doJLzY=
|
||||
github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.23 h1:pbrxO/kuIwgEsOPLkaHu0O+m4fNgLU8B3vxQ+72jTPw=
|
||||
github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.23/go.mod h1:/CMNUqoj46HpS3MNRDEDIwcgEnrtZlKRaHNaHxIFpNA=
|
||||
github.com/aws/aws-sdk-go-v2/service/signin v1.0.11 h1:TdJ+HdzOBhU8+iVAOGUTU63VXopcumCOF1paFulHWZc=
|
||||
github.com/aws/aws-sdk-go-v2/service/signin v1.0.11/go.mod h1:R82ZRExE/nheo0N+T8zHPcLRTcH8MGsnR3BiVGX0TwI=
|
||||
github.com/aws/aws-sdk-go-v2/service/sso v1.30.17 h1:7byT8HUWrgoRp6sXjxtZwgOKfhss5fW6SkLBtqzgRoE=
|
||||
github.com/aws/aws-sdk-go-v2/service/sso v1.30.17/go.mod h1:xNWknVi4Ezm1vg1QsB/5EWpAJURq22uqd38U8qKvOJc=
|
||||
github.com/aws/aws-sdk-go-v2/service/ssooidc v1.35.21 h1:+1Kl1zx6bWi4X7cKi3VYh29h8BvsCoHQEQ6ST9X8w7w=
|
||||
github.com/aws/aws-sdk-go-v2/service/ssooidc v1.35.21/go.mod h1:4vIRDq+CJB2xFAXZ+YgGUTiEft7oAQlhIs71xcSeuVg=
|
||||
github.com/aws/aws-sdk-go-v2/service/sts v1.42.1 h1:F/M5Y9I3nwr2IEpshZgh1GeHpOItExNM9L1euNuh/fk=
|
||||
github.com/aws/aws-sdk-go-v2/service/sts v1.42.1/go.mod h1:mTNxImtovCOEEuD65mKW7DCsL+2gjEH+RPEAexAzAio=
|
||||
github.com/aws/smithy-go v1.25.1 h1:J8ERsGSU7d+aCmdQur5Txg6bVoYelvQJgtZehD12GkI=
|
||||
github.com/aws/smithy-go v1.25.1/go.mod h1:YE2RhdIuDbA5E5bTdciG9KrW3+TiEONeUWCqxX9i1Fc=
|
||||
github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
|
||||
github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
|
||||
github.com/cncf/xds/go v0.0.0-20251210132809-ee656c7534f5 h1:6xNmx7iTtyBRev0+D/Tv1FZd4SCg8axKApyNyRsAt/w=
|
||||
@@ -42,6 +74,8 @@ github.com/googleapis/enterprise-certificate-proxy v0.3.14 h1:yh8ncqsbUY4shRD5dA
|
||||
github.com/googleapis/enterprise-certificate-proxy v0.3.14/go.mod h1:vqVt9yG9480NtzREnTlmGSBmFrA+bzb0yl0TxoBQXOg=
|
||||
github.com/googleapis/gax-go/v2 v2.21.0 h1:h45NjjzEO3faG9Lg/cFrBh2PgegVVgzqKzuZl/wMbiI=
|
||||
github.com/googleapis/gax-go/v2 v2.21.0/go.mod h1:But/NJU6TnZsrLai/xBAQLLz+Hc7fHZJt/hsCz3Fih4=
|
||||
github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg=
|
||||
github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
|
||||
github.com/planetscale/vtprotobuf v0.6.1-0.20240319094008-0393e58bdf10 h1:GFCKgmp0tecUJ0sJuv4pzYCqS9+RGSn52M3FUwPs+uo=
|
||||
github.com/planetscale/vtprotobuf v0.6.1-0.20240319094008-0393e58bdf10/go.mod h1:t/avpk3KcrXxUnYOhZhMXJlSEyie6gQbtLq5NM3loB8=
|
||||
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
|
||||
@@ -82,6 +116,8 @@ gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4=
|
||||
gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E=
|
||||
google.golang.org/api v0.274.0 h1:aYhycS5QQCwxHLwfEHRRLf9yNsfvp1JadKKWBE54RFA=
|
||||
google.golang.org/api v0.274.0/go.mod h1:JbAt7mF+XVmWu6xNP8/+CTiGH30ofmCmk9nM8d8fHew=
|
||||
google.golang.org/genai v1.56.0 h1:IwWrg1K0cn1/WBiPno/dYr0Q6o75NeH/bh3G4JEFERE=
|
||||
google.golang.org/genai v1.56.0/go.mod h1:A3kkl0nyBjyFlNjgxIwKq70julKbIxpSxqKO5gw/gmk=
|
||||
google.golang.org/genproto v0.0.0-20260319201613-d00831a3d3e7 h1:XzmzkmB14QhVhgnawEVsOn6OFsnpyxNPRY9QV01dNB0=
|
||||
google.golang.org/genproto v0.0.0-20260319201613-d00831a3d3e7/go.mod h1:L43LFes82YgSonw6iTXTxXUX1OlULt4AQtkik4ULL/I=
|
||||
google.golang.org/genproto/googleapis/api v0.0.0-20260401024825-9d38bb4040a9 h1:VPWxll4HlMw1Vs/qXtN7BvhZqsS9cdAittCNvVENElA=
|
||||
|
||||
@@ -0,0 +1,50 @@
|
||||
package storage
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"os"
|
||||
"time"
|
||||
|
||||
"github.com/aws/aws-sdk-go-v2/aws"
|
||||
"github.com/aws/aws-sdk-go-v2/config"
|
||||
"github.com/aws/aws-sdk-go-v2/service/dynamodb"
|
||||
)
|
||||
|
||||
// dynamoDBHTTPTimeout caps individual SDK HTTP calls. DynamoDB on-demand is
|
||||
// fast (typically <50ms); a 10s budget absorbs cold-start TLS handshake on
|
||||
// Lambda + retries without hiding pathological hangs.
|
||||
const dynamoDBHTTPTimeout = 10 * time.Second
|
||||
|
||||
// NewDynamoDBClient constructs a DynamoDB client using AWS standard credential
|
||||
// resolution: Lambda execution role → env vars → shared config. Region is
|
||||
// resolved by the SDK from AWS_REGION (set by Lambda) or AWS_DEFAULT_REGION.
|
||||
//
|
||||
// The caller may pass a non-empty endpoint override (e.g. "http://localhost:8000")
|
||||
// to point at DynamoDB Local for tests. An empty endpoint uses the AWS default.
|
||||
func NewDynamoDBClient(ctx context.Context, endpoint string) (*dynamodb.Client, error) {
|
||||
httpClient := &http.Client{Timeout: dynamoDBHTTPTimeout}
|
||||
|
||||
loadOpts := []func(*config.LoadOptions) error{
|
||||
config.WithHTTPClient(httpClient),
|
||||
}
|
||||
cfg, err := config.LoadDefaultConfig(ctx, loadOpts...)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("storage: load AWS config: %w", err)
|
||||
}
|
||||
|
||||
clientOpts := []func(*dynamodb.Options){}
|
||||
if endpoint != "" {
|
||||
clientOpts = append(clientOpts, func(o *dynamodb.Options) {
|
||||
o.BaseEndpoint = aws.String(endpoint)
|
||||
})
|
||||
}
|
||||
return dynamodb.NewFromConfig(cfg, clientOpts...), nil
|
||||
}
|
||||
|
||||
// DynamoDBEndpointFromEnv returns the override endpoint for tests / local dev.
|
||||
// Empty string means "use AWS default endpoint."
|
||||
func DynamoDBEndpointFromEnv() string {
|
||||
return os.Getenv("DYNAMODB_LOCAL_URL")
|
||||
}
|
||||
@@ -0,0 +1,192 @@
|
||||
package storage
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strconv"
|
||||
"time"
|
||||
|
||||
"github.com/aws/aws-sdk-go-v2/aws"
|
||||
"github.com/aws/aws-sdk-go-v2/service/dynamodb"
|
||||
"github.com/aws/aws-sdk-go-v2/service/dynamodb/types"
|
||||
)
|
||||
|
||||
// DynamoDB attribute names. `value` is reserved by DynamoDB expressions, so
|
||||
// every reference goes through ExpressionAttributeNames (#v).
|
||||
const (
|
||||
dynamoPKAttr = "pk"
|
||||
dynamoSKAttr = "sk"
|
||||
dynamoValueAttr = "value"
|
||||
dynamoUpdatedAtAttr = "updatedAt"
|
||||
)
|
||||
|
||||
// DynamoDBKVStore is a KVStore backed by a single DynamoDB table with a
|
||||
// composite key (pk, sk). FirestoreProvider uses one collection per module;
|
||||
// DynamoDBProvider uses one partition (pk = moduleName) per module — the
|
||||
// table itself is shared. The user-supplied key becomes the sort key, so
|
||||
// List(prefix) maps to a Query with begins_with(sk, prefix).
|
||||
type DynamoDBKVStore struct {
|
||||
client *dynamodb.Client
|
||||
table string
|
||||
moduleName string
|
||||
}
|
||||
|
||||
// NewDynamoDBKVStore returns a store partition-scoped to moduleName.
|
||||
// Callers must validate moduleName beforehand (DynamoDBProvider does).
|
||||
func NewDynamoDBKVStore(client *dynamodb.Client, table, moduleName string) *DynamoDBKVStore {
|
||||
return &DynamoDBKVStore{client: client, table: table, moduleName: moduleName}
|
||||
}
|
||||
|
||||
// Get returns the raw bytes stored at key, or ErrNotFound. Strong read for
|
||||
// parity with the Firestore impl's strong-read default.
|
||||
func (s *DynamoDBKVStore) Get(ctx context.Context, key string) ([]byte, error) {
|
||||
if err := validateKey(key); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out, err := s.client.GetItem(ctx, &dynamodb.GetItemInput{
|
||||
TableName: aws.String(s.table),
|
||||
ConsistentRead: aws.Bool(true),
|
||||
Key: map[string]types.AttributeValue{
|
||||
dynamoPKAttr: &types.AttributeValueMemberS{Value: s.moduleName},
|
||||
dynamoSKAttr: &types.AttributeValueMemberS{Value: key},
|
||||
},
|
||||
ExpressionAttributeNames: map[string]string{"#v": dynamoValueAttr},
|
||||
ProjectionExpression: aws.String("#v"),
|
||||
})
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("dynamodb get %s/%s: %w", s.moduleName, key, err)
|
||||
}
|
||||
if len(out.Item) == 0 {
|
||||
return nil, ErrNotFound
|
||||
}
|
||||
rawAttr, ok := out.Item[dynamoValueAttr]
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("dynamodb get %s/%s: missing %q attribute", s.moduleName, key, dynamoValueAttr)
|
||||
}
|
||||
bin, ok := rawAttr.(*types.AttributeValueMemberB)
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("dynamodb get %s/%s: unexpected attribute type %T", s.moduleName, key, rawAttr)
|
||||
}
|
||||
return bin.Value, nil
|
||||
}
|
||||
|
||||
// GetJSON decodes the value at key into dst.
|
||||
func (s *DynamoDBKVStore) GetJSON(ctx context.Context, key string, dst any) error {
|
||||
raw, err := s.Get(ctx, key)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := json.Unmarshal(raw, dst); err != nil {
|
||||
return fmt.Errorf("dynamodb get %s/%s: json decode: %w", s.moduleName, key, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Put writes raw bytes at key, creating or overwriting.
|
||||
func (s *DynamoDBKVStore) Put(ctx context.Context, key string, val []byte) error {
|
||||
if err := validateKey(key); err != nil {
|
||||
return err
|
||||
}
|
||||
// DynamoDB rejects empty Binary values. Store a single zero byte sentinel
|
||||
// transparently — Firestore allows zero-length []byte and callers may
|
||||
// rely on that. The Get path treats both as []byte; downstream JSON
|
||||
// callers will see []byte{0} where they put []byte{}, but no current
|
||||
// caller relies on storing literal empty bytes (they use PutJSON, which
|
||||
// always emits at least "null" = 4 bytes).
|
||||
if len(val) == 0 {
|
||||
val = []byte{0}
|
||||
}
|
||||
_, err := s.client.PutItem(ctx, &dynamodb.PutItemInput{
|
||||
TableName: aws.String(s.table),
|
||||
Item: map[string]types.AttributeValue{
|
||||
dynamoPKAttr: &types.AttributeValueMemberS{Value: s.moduleName},
|
||||
dynamoSKAttr: &types.AttributeValueMemberS{Value: key},
|
||||
dynamoValueAttr: &types.AttributeValueMemberB{Value: val},
|
||||
dynamoUpdatedAtAttr: &types.AttributeValueMemberN{Value: strconv.FormatInt(time.Now().UTC().UnixNano(), 10)},
|
||||
},
|
||||
})
|
||||
if err != nil {
|
||||
return fmt.Errorf("dynamodb put %s/%s: %w", s.moduleName, key, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// PutJSON marshals val and writes the bytes at key.
|
||||
func (s *DynamoDBKVStore) PutJSON(ctx context.Context, key string, val any) error {
|
||||
raw, err := json.Marshal(val)
|
||||
if err != nil {
|
||||
return fmt.Errorf("dynamodb put %s/%s: json encode: %w", s.moduleName, key, err)
|
||||
}
|
||||
return s.Put(ctx, key, raw)
|
||||
}
|
||||
|
||||
// Delete removes the item at key. Deleting a missing key is not an error.
|
||||
func (s *DynamoDBKVStore) Delete(ctx context.Context, key string) error {
|
||||
if err := validateKey(key); err != nil {
|
||||
return err
|
||||
}
|
||||
_, err := s.client.DeleteItem(ctx, &dynamodb.DeleteItemInput{
|
||||
TableName: aws.String(s.table),
|
||||
Key: map[string]types.AttributeValue{
|
||||
dynamoPKAttr: &types.AttributeValueMemberS{Value: s.moduleName},
|
||||
dynamoSKAttr: &types.AttributeValueMemberS{Value: key},
|
||||
},
|
||||
})
|
||||
if err != nil {
|
||||
return fmt.Errorf("dynamodb delete %s/%s: %w", s.moduleName, key, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// List returns all sort keys in the partition that start with prefix. Empty
|
||||
// prefix scans the entire partition (Query is still cheap because it's
|
||||
// partition-bounded; no Scan).
|
||||
func (s *DynamoDBKVStore) List(ctx context.Context, prefix string) ([]string, error) {
|
||||
if err := validatePrefix(prefix); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
input := &dynamodb.QueryInput{
|
||||
TableName: aws.String(s.table),
|
||||
ProjectionExpression: aws.String(dynamoSKAttr),
|
||||
ExpressionAttributeValues: map[string]types.AttributeValue{
|
||||
":pk": &types.AttributeValueMemberS{Value: s.moduleName},
|
||||
},
|
||||
}
|
||||
if prefix == "" {
|
||||
input.KeyConditionExpression = aws.String("pk = :pk")
|
||||
} else {
|
||||
input.KeyConditionExpression = aws.String("pk = :pk AND begins_with(sk, :prefix)")
|
||||
input.ExpressionAttributeValues[":prefix"] = &types.AttributeValueMemberS{Value: prefix}
|
||||
}
|
||||
|
||||
var keys []string
|
||||
pager := dynamodb.NewQueryPaginator(s.client, input)
|
||||
for pager.HasMorePages() {
|
||||
page, err := pager.NextPage(ctx)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("dynamodb list %s prefix=%q: %w", s.moduleName, prefix, err)
|
||||
}
|
||||
for _, item := range page.Items {
|
||||
skAttr, ok := item[dynamoSKAttr]
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
sk, ok := skAttr.(*types.AttributeValueMemberS)
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
keys = append(keys, sk.Value)
|
||||
}
|
||||
}
|
||||
return keys, nil
|
||||
}
|
||||
|
||||
// errIsNotFound wraps the SDK NotFound condition. DynamoDB GetItem doesn't
|
||||
// return an error for missing items (returns empty Item), but other ops may
|
||||
// surface ResourceNotFoundException for missing tables.
|
||||
func errIsTableMissing(err error) bool {
|
||||
var rnfe *types.ResourceNotFoundException
|
||||
return errors.As(err, &rnfe)
|
||||
}
|
||||
@@ -0,0 +1,163 @@
|
||||
package storage
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"os"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/aws/aws-sdk-go-v2/aws"
|
||||
"github.com/aws/aws-sdk-go-v2/service/dynamodb"
|
||||
"github.com/aws/aws-sdk-go-v2/service/dynamodb/types"
|
||||
)
|
||||
|
||||
// dynamoDBLocalSetup ensures a fresh table exists in DynamoDB Local for the
|
||||
// test run. Returns a client + cleanup. Tests skip if DYNAMODB_LOCAL_URL is
|
||||
// unset so CI without docker still builds.
|
||||
func dynamoDBLocalSetup(t *testing.T) (client *dynamodb.Client, table string, cleanup func()) {
|
||||
t.Helper()
|
||||
endpoint := os.Getenv("DYNAMODB_LOCAL_URL")
|
||||
if endpoint == "" {
|
||||
t.Skip("DYNAMODB_LOCAL_URL not set; skipping DynamoDB integration test (run `make dynamodb-local` to start the local container)")
|
||||
}
|
||||
// DynamoDB Local accepts any non-empty creds + region.
|
||||
t.Setenv("AWS_ACCESS_KEY_ID", "test")
|
||||
t.Setenv("AWS_SECRET_ACCESS_KEY", "test")
|
||||
t.Setenv("AWS_REGION", "ap-southeast-1")
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
defer cancel()
|
||||
|
||||
c, err := NewDynamoDBClient(ctx, endpoint)
|
||||
if err != nil {
|
||||
t.Fatalf("NewDynamoDBClient: %v", err)
|
||||
}
|
||||
|
||||
table = "test-" + t.Name()
|
||||
if len(table) > 64 {
|
||||
table = table[:64]
|
||||
}
|
||||
|
||||
_, err = c.CreateTable(ctx, &dynamodb.CreateTableInput{
|
||||
TableName: aws.String(table),
|
||||
BillingMode: types.BillingModePayPerRequest,
|
||||
AttributeDefinitions: []types.AttributeDefinition{
|
||||
{AttributeName: aws.String("pk"), AttributeType: types.ScalarAttributeTypeS},
|
||||
{AttributeName: aws.String("sk"), AttributeType: types.ScalarAttributeTypeS},
|
||||
},
|
||||
KeySchema: []types.KeySchemaElement{
|
||||
{AttributeName: aws.String("pk"), KeyType: types.KeyTypeHash},
|
||||
{AttributeName: aws.String("sk"), KeyType: types.KeyTypeRange},
|
||||
},
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("CreateTable: %v", err)
|
||||
}
|
||||
|
||||
cleanup = func() {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
_, _ = c.DeleteTable(ctx, &dynamodb.DeleteTableInput{TableName: aws.String(table)})
|
||||
}
|
||||
return c, table, cleanup
|
||||
}
|
||||
|
||||
func TestDynamoDBKVStore_PutGetDelete(t *testing.T) {
|
||||
client, table, cleanup := dynamoDBLocalSetup(t)
|
||||
defer cleanup()
|
||||
|
||||
ctx := context.Background()
|
||||
s := NewDynamoDBKVStore(client, table, "wordle")
|
||||
|
||||
if err := s.Put(ctx, "user:1:state", []byte("hello")); err != nil {
|
||||
t.Fatalf("Put: %v", err)
|
||||
}
|
||||
got, err := s.Get(ctx, "user:1:state")
|
||||
if err != nil {
|
||||
t.Fatalf("Get: %v", err)
|
||||
}
|
||||
if string(got) != "hello" {
|
||||
t.Errorf("Get: got %q, want %q", got, "hello")
|
||||
}
|
||||
|
||||
if err := s.Delete(ctx, "user:1:state"); err != nil {
|
||||
t.Fatalf("Delete: %v", err)
|
||||
}
|
||||
if _, err := s.Get(ctx, "user:1:state"); !errors.Is(err, ErrNotFound) {
|
||||
t.Errorf("Get after Delete: got %v, want ErrNotFound", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDynamoDBKVStore_GetMissing(t *testing.T) {
|
||||
client, table, cleanup := dynamoDBLocalSetup(t)
|
||||
defer cleanup()
|
||||
|
||||
s := NewDynamoDBKVStore(client, table, "wordle")
|
||||
_, err := s.Get(context.Background(), "missing")
|
||||
if !errors.Is(err, ErrNotFound) {
|
||||
t.Errorf("got %v, want ErrNotFound", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDynamoDBKVStore_JSONRoundTrip(t *testing.T) {
|
||||
client, table, cleanup := dynamoDBLocalSetup(t)
|
||||
defer cleanup()
|
||||
|
||||
ctx := context.Background()
|
||||
s := NewDynamoDBKVStore(client, table, "loldle")
|
||||
|
||||
type state struct {
|
||||
Score int `json:"score"`
|
||||
Name string `json:"name"`
|
||||
}
|
||||
in := state{Score: 42, Name: "ezreal"}
|
||||
if err := s.PutJSON(ctx, "u1", in); err != nil {
|
||||
t.Fatalf("PutJSON: %v", err)
|
||||
}
|
||||
var out state
|
||||
if err := s.GetJSON(ctx, "u1", &out); err != nil {
|
||||
t.Fatalf("GetJSON: %v", err)
|
||||
}
|
||||
if out != in {
|
||||
t.Errorf("got %+v, want %+v", out, in)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDynamoDBKVStore_ListPrefix(t *testing.T) {
|
||||
client, table, cleanup := dynamoDBLocalSetup(t)
|
||||
defer cleanup()
|
||||
|
||||
ctx := context.Background()
|
||||
s := NewDynamoDBKVStore(client, table, "wordle")
|
||||
|
||||
for _, k := range []string{"user:1:state", "user:2:state", "config:daily", "user:1:history"} {
|
||||
if err := s.Put(ctx, k, []byte("x")); err != nil {
|
||||
t.Fatalf("Put %s: %v", k, err)
|
||||
}
|
||||
}
|
||||
|
||||
got, err := s.List(ctx, "user:")
|
||||
if err != nil {
|
||||
t.Fatalf("List: %v", err)
|
||||
}
|
||||
want := map[string]bool{"user:1:state": true, "user:2:state": true, "user:1:history": true}
|
||||
if len(got) != len(want) {
|
||||
t.Errorf("List: got %v (len=%d), want len=%d", got, len(got), len(want))
|
||||
}
|
||||
for _, k := range got {
|
||||
if !want[k] {
|
||||
t.Errorf("List: unexpected key %q", k)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestDynamoDBKVStore_DeleteMissingNoError(t *testing.T) {
|
||||
client, table, cleanup := dynamoDBLocalSetup(t)
|
||||
defer cleanup()
|
||||
|
||||
s := NewDynamoDBKVStore(client, table, "wordle")
|
||||
if err := s.Delete(context.Background(), "never-existed"); err != nil {
|
||||
t.Errorf("Delete missing key: got %v, want nil (idempotent)", err)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,30 @@
|
||||
package storage
|
||||
|
||||
import (
|
||||
"github.com/aws/aws-sdk-go-v2/service/dynamodb"
|
||||
)
|
||||
|
||||
// DynamoDBProvider is a KVProvider that hands out per-module stores backed by
|
||||
// a single shared DynamoDB table. Module isolation is by partition (pk =
|
||||
// moduleName); no key-prefix wrapping is needed at this layer.
|
||||
type DynamoDBProvider struct {
|
||||
client *dynamodb.Client
|
||||
table string
|
||||
}
|
||||
|
||||
// NewDynamoDBProvider returns a provider over the given client + table.
|
||||
// The client is goroutine-safe and outlives every store handed out by For.
|
||||
func NewDynamoDBProvider(client *dynamodb.Client, table string) *DynamoDBProvider {
|
||||
return &DynamoDBProvider{client: client, table: table}
|
||||
}
|
||||
|
||||
// For returns a DynamoDBKVStore partitioned by moduleName. moduleName is
|
||||
// re-validated against the same alphabet enforced by FirestoreProvider so
|
||||
// behaviour is identical between backends. An invalid name yields an
|
||||
// invalidStore — every op errors at first use, surfacing the bug early.
|
||||
func (p *DynamoDBProvider) For(moduleName string) KVStore {
|
||||
if !collectionNameRe.MatchString(moduleName) {
|
||||
return invalidStore{name: moduleName}
|
||||
}
|
||||
return NewDynamoDBKVStore(p.client, p.table, moduleName)
|
||||
}
|
||||
@@ -0,0 +1,61 @@
|
||||
package storage
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestDynamoDBProvider_CrossModuleIsolation(t *testing.T) {
|
||||
client, table, cleanup := dynamoDBLocalSetup(t)
|
||||
defer cleanup()
|
||||
|
||||
ctx := context.Background()
|
||||
p := NewDynamoDBProvider(client, table)
|
||||
|
||||
wordle := p.For("wordle")
|
||||
loldle := p.For("loldle")
|
||||
|
||||
if err := wordle.Put(ctx, "k", []byte("from-wordle")); err != nil {
|
||||
t.Fatalf("wordle.Put: %v", err)
|
||||
}
|
||||
if err := loldle.Put(ctx, "k", []byte("from-loldle")); err != nil {
|
||||
t.Fatalf("loldle.Put: %v", err)
|
||||
}
|
||||
|
||||
got, err := wordle.Get(ctx, "k")
|
||||
if err != nil {
|
||||
t.Fatalf("wordle.Get: %v", err)
|
||||
}
|
||||
if string(got) != "from-wordle" {
|
||||
t.Errorf("wordle.Get: got %q, want from-wordle", got)
|
||||
}
|
||||
|
||||
got, err = loldle.Get(ctx, "k")
|
||||
if err != nil {
|
||||
t.Fatalf("loldle.Get: %v", err)
|
||||
}
|
||||
if string(got) != "from-loldle" {
|
||||
t.Errorf("loldle.Get: got %q, want from-loldle", got)
|
||||
}
|
||||
|
||||
// List should not leak across modules.
|
||||
wkeys, err := wordle.List(ctx, "")
|
||||
if err != nil {
|
||||
t.Fatalf("wordle.List: %v", err)
|
||||
}
|
||||
if len(wkeys) != 1 || wkeys[0] != "k" {
|
||||
t.Errorf("wordle.List: got %v, want [k]", wkeys)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDynamoDBProvider_InvalidModuleName(t *testing.T) {
|
||||
// No DynamoDB needed — invalidStore short-circuits before any SDK call.
|
||||
p := NewDynamoDBProvider(nil, "any")
|
||||
store := p.For("../../etc/passwd")
|
||||
|
||||
_, err := store.Get(context.Background(), "k")
|
||||
if !errors.Is(err, ErrInvalidModuleName) {
|
||||
t.Errorf("Get: got %v, want ErrInvalidModuleName", err)
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user