Global ttlInSeconds metadata configuration for DynamoDb

Signed-off-by: Zeynel Koca <[email protected]>

Author: ZeynelKoca

state/aws/dynamodb/dynamodb.go

@@ -48,6 +48,7 @@ type StateStore struct {
 	table            string
 	ttlAttributeName string
 	partitionKey     string
+	ttlInSeconds     *int
 
 	dynamodbClient *dynamodb.Client
 }
@@ -64,6 +65,7 @@ type dynamoDBMetadata struct {
 	Table            string `json:"table"`
 	TTLAttributeName string `json:"ttlAttributeName"`
 	PartitionKey     string `json:"partitionKey"`
+	TTLInSeconds     *int   `json:"ttlInSeconds" mapstructure:"ttlInSeconds"`
 }
 
 type putData struct {
@@ -117,6 +119,7 @@ func (d *StateStore) Init(ctx context.Context, metadata state.Metadata) error {
 	d.table = meta.Table
 	d.ttlAttributeName = meta.TTLAttributeName
 	d.partitionKey = meta.PartitionKey
+	d.ttlInSeconds = meta.TTLInSeconds
 
 	if err := d.validateTableAccess(ctx); err != nil {
 		return fmt.Errorf("error validating DynamoDB table '%s' access: %w", d.table, err)
@@ -425,6 +428,11 @@ func (d *StateStore) parseTTL(req *state.SetRequest) (*int64, error) {
 
 			return &expirationTime, nil
 		}
+		// apply global TTL if no explicit TTL in request metadata
+		if d.ttlInSeconds != nil {
+			expirationTime := time.Now().Unix() + int64(*d.ttlInSeconds)
+			return &expirationTime, nil
+		}
 	}
 
 	return nil, nil
@@ -471,7 +479,26 @@ func (d *StateStore) Multi(ctx context.Context, request *state.TransactionalStat
 			if err != nil {
 				return fmt.Errorf("dynamodb error: failed to marshal value for key %s: %w", req.Key, err)
 			}
-			twi.Put = pd.ToPut()
+			ttl, err := d.parseTTL(&req)
+			if err != nil {
+				return fmt.Errorf("dynamodb error: failed to parse ttlInSeconds: %w", err)
+			}
+			twi.Put = &types.Put{
+				TableName: ptr.Of(d.table),
+				Item: map[string]types.AttributeValue{
+					d.partitionKey: &types.AttributeValueMemberS{
+						Value: req.Key,
+					},
+					"value": &types.AttributeValueMemberS{
+						Value: value,
+					},
+				},
+			}
+			if ttl != nil {
+				twi.Put.Item[d.ttlAttributeName] = &types.AttributeValueMemberN{
+					Value: strconv.FormatInt(*ttl, 10),
+				}
+			}
 
 		case state.DeleteRequest:
 			twi.Delete = &types.Delete{