Day 7 — DynamoDB and Caching

flowchart TB
    K["Item key:<br/>customerId = c-51<br/>orderId = o-8812"] --> H["hash(c-51)"]
    H --> P{"Which partition?"}
    P --> P1["Partition 1"]
    P --> P2["Partition 2<br/>all items for c-51 live here<br/>o-1001<br/>o-4520<br/>o-8812 (sorted by sort key)<br/>o-9903"]
    P --> P3["Partition 3"]
    P2 --> R["Read one item: O(1)<br/>Read a RANGE of one customer's<br/>orders: O(result size)"]

RELATIONAL   index on (customer_id, created_at)
             - traverse a B-tree, seek to the customer, read in order
             - cost grows (slowly) with total data size
             - and you may add a different index tomorrow for a new query

DYNAMODB     partition key = customer_id, sort key = created_at
             - hash straight to the partition, read in order
             - cost is independent of total data size
             - but you CANNOT add a different access pattern tomorrow
               without designing for it (a GSI, or a rewrite)

6 KB rounds up to 2 x 4 KB units
strongly consistent:   500 x 2      = 1,000 RCU
eventually consistent: 500 x 2 / 2  =   500 RCU   <- half the cost, for a read
                                                     that may be a few ms stale

3 KB rounds up to 3 x 1 KB units
200 x 3 = 600 WCU

1,000 WCU steady for a month:
  provisioned:  1,000 x $0.00013 x 730 h            = about  $95
  on-demand:    1,000 x 3600 x 730 / 1M x $1.25     = about $3,285   (34x more)

TABLE: orders          PK = pk (String)        SK = sk (String)

Pattern 1 and 6 - the order and its items are ONE contiguous read:
  pk = ORDER#o-8812    sk = META                 -> the order itself
  pk = ORDER#o-8812    sk = ITEM#001             |
  pk = ORDER#o-8812    sk = ITEM#002             |  Query pk=ORDER#o-8812
  pk = ORDER#o-8812    sk = ITEM#003             |  returns all of it, once

Pattern 2 - GSI1, newest first:
  GSI1PK = CUST#c-51   GSI1SK = 2026-01-14T09:12:04Z#o-8812
  -> Query GSI1 with pk=CUST#c-51, ScanIndexForward=false, Limit=20

Pattern 3 - GSI2, status within a customer:
  GSI2PK = CUST#c-51   GSI2SK = PAID#2026-01-14T09:12:04Z
  -> Query GSI2 with pk=CUST#c-51 AND begins_with(sk, "PAID#")

Pattern 4 - GSI3, SPARSE: the attribute exists ONLY while pending.
  GSI3PK = PENDING     GSI3SK = 2026-01-14T09:12:04Z
  -> when the order is paid, DELETE the GSI3PK attribute; the item leaves
     the index entirely. The index stays small forever.
     But see the hot-partition discussion: a single PK value is one partition.

Pattern 5 - GSI4, with a write-sharded key to avoid a daily hot partition:
  GSI4PK = DATE#2026-01-14#<0-9>    GSI4SK = <timestamp>#o-8812
  -> query all 10 shards in parallel and merge. Low frequency, so the
     extra complexity buys throughput where a single key would throttle.

// Create only if it does not exist. The atomic primitive behind idempotency.
.conditionExpression("attribute_not_exists(pk)")

// Optimistic locking - the DynamoDB equivalent of a version column.
.conditionExpression("#v = :expectedVersion")

// Business invariant enforced in the database, not in application logic.
.conditionExpression("#status = :pending AND #stock >= :qty")

flowchart TB
    subgraph CA["Cache-aside (lazy loading) - the default"]
        A1[Read] --> A2{"in cache?"}
        A2 -->|hit| A3[return]
        A2 -->|miss| A4[read DB] --> A5[write cache] --> A3
        A6[Write] --> A7[write DB] --> A8["invalidate cache"]
    end
    subgraph WT["Write-through"]
        B1[Write] --> B2[write cache] --> B3[write DB]
        B4[Read] --> B5["always a hit for written data"]
    end
    subgraph WB["Write-behind"]
        C1[Write] --> C2[write cache] --> C3["return immediately"]
        C2 -.->|async, batched| C4[write DB]
    end

flowchart LR
    K["Hot key TTL expires"] --> M["1,000 concurrent misses"]
    M --> DB[("Database gets 1,000x<br/>its normal load, instantly")]
    DB --> T["Timeouts, then retries, then worse"]

// INCR + EXPIRE on a time-bucketed key. Atomic, so no race between
// checking and incrementing.
String key = "rate:" + userId + ":" + (System.currentTimeMillis() / 60_000);
Long count = redis.opsForValue().increment(key);
if (count == 1L) redis.expire(key, Duration.ofMinutes(2));
if (count > LIMIT) throw new TooManyRequestsException();

// SET key value NX PX ttl  - atomic acquire with an expiry
Boolean acquired = redis.opsForValue()
        .setIfAbsent(lockKey, ownerToken, Duration.ofSeconds(30));

flowchart TB
    C[Client] --> ALB[ALB] --> APP[Order Service · ECS]
    subgraph DATA["Data tier"]
        RED[("ElastiCache Redis<br/>cluster mode enabled<br/>catalogue cache")]
        DDB[("DynamoDB<br/>orders + GSIs")]
        IDEM[("DynamoDB<br/>idempotency keys · TTL 24h")]
        AUR[("Aurora<br/>system of record")]
    end
    APP -->|"1 cache-aside read"| RED
    RED -.->|"miss"| AUR
    APP -->|"high-volume order reads/writes"| DDB
    APP -->|"conditional write before side effects"| IDEM
    DDB -->|"Streams to Day 9"| STR[(DynamoDB Streams)]

sequenceDiagram
    autonumber
    participant APP as Order Service
    participant R as Redis
    participant D as DynamoDB
    APP->>R: GET catalogue:sku-991
    alt hit (about 0.3 ms)
        R-->>APP: value
    else miss
        R-->>APP: nil
        APP->>APP: acquire per-key in-process lock<br/>(request coalescing, stops a stampede)
        APP->>D: GetItem pk=SKU#991 (eventually consistent, 0.5 RCU)
        D-->>APP: item (about 4 ms)
        APP->>R: SETEX catalogue:sku-991 with jittered TTL
        APP->>APP: release lock; waiters read the cached value
    end

flowchart TB
    APP[Application] --> R{"Redis reachable?"}
    R -->|"no, node failure"| FB["FALL BACK to DynamoDB.<br/>A cache outage must degrade<br/>latency and cost, NEVER<br/>availability."]
    FB --> HERD["but now 100% of reads hit the DB:<br/>thundering herd"]
    HERD --> CB["Circuit breaker + load shedding<br/>keep the DB alive (Day 12)"]
    R -->|yes| HIT[serve]
    APP --> D{"DynamoDB throttles?"}
    D -->|"ProvisionedThroughputExceeded"| RETRY["SDK retries with backoff<br/>(automatic in v2)"]
    RETRY -->|"persistent"| HOT["Hot partition or<br/>under-provisioned GSI:<br/>a KEY DESIGN problem,<br/>not a capacity problem"]
    D -->|"GSI throttled"| BASE["BASE TABLE WRITES<br/>are throttled too"]

sequenceDiagram
    autonumber
    participant APP as Application
    participant EP as dynamodb.eu-west-1.amazonaws.com
    participant RT as Request router
    participant PART as Storage partition (3 replicas across AZs)
    APP->>EP: GetItem pk=ORDER#o-8812 (SigV4 signed)
    EP->>RT: authorize (IAM), then route
    RT->>RT: hash(pk) gives partition id and replica set
    alt Eventually consistent read (default)
        RT->>PART: read from ANY of the 3 replicas
        Note over PART: may miss a write from the<br/>last few milliseconds. 0.5 RCU.
    else Strongly consistent read
        RT->>PART: read from the LEADER replica
        Note over PART: reflects all prior writes. 1 RCU.
    end
    PART-->>APP: item + ConsumedCapacity

sequenceDiagram
    autonumber
    participant APP
    participant RT as Request router
    participant L as Leader replica
    participant F as Follower replicas (2, other AZs)
    APP->>RT: PutItem with ConditionExpression
    RT->>L: route to the leader for this partition
    L->>L: evaluate the condition ATOMICALLY on the item
    alt condition fails
        L-->>APP: ConditionalCheckFailedException<br/>(capacity IS still consumed)
    else condition holds
        L->>F: replicate
        F-->>L: ack (durability quorum)
        L-->>APP: 200 + ConsumedCapacity
        L-->>APP: Streams record written
        L-->>APP: GSI update is asynchronous and<br/>consumes the GSI's own capacity
    end

flowchart TB
    subgraph BAD["BAD: pk = status"]
        B1["hash(PENDING)"] --> BP1["Partition A<br/>95% of all traffic<br/>ceiling: 1,000 WCU"]
        B2["hash(SHIPPED)"] --> BP2["Partition B<br/>nearly idle"]
        BP1 --> THR["THROTTLED at about 1,000 WCU<br/>even with 20,000 provisioned:<br/>the other 19,000 are unreachable"]
    end
    subgraph GOOD["GOOD: pk = orderId"]
        G1["hash(o-8812)"] --> GP1[Partition 1]
        G2["hash(o-8813)"] --> GP2[Partition 2]
        G3["hash(o-9001)"] --> GP3[Partition N]
        GP1 & GP2 & GP3 --> OK["Even distribution.<br/>Scales to the table's full capacity."]
    end

<dependency>
  <groupId>software.amazon.awssdk</groupId>
  <artifactId>dynamodb-enhanced</artifactId>   <!-- BOM-managed -->
</dependency>
<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>

package com.acme.orders;

import software.amazon.awssdk.enhanced.dynamodb.mapper.annotations.*;

import java.math.BigDecimal;
import java.time.Instant;

@DynamoDbBean
public class OrderItem {

    private String pk;          // ORDER#o-8812
    private String sk;          // META, or ITEM#001
    private String orderId;
    private String customerId;
    private String status;
    private BigDecimal amount;
    private Instant createdAt;
    private Long version;

    // GSI1: a customer's orders, newest first
    private String gsi1pk;      // CUST#c-51
    private String gsi1sk;      // 2026-01-14T09:12:04Z#o-8812

    // GSI3: SPARSE work queue. These attributes exist ONLY while pending;
    // clearing them removes the item from the index entirely.
    private String gsi3pk;      // PENDING#<shard>, or null
    private String gsi3sk;

    @DynamoDbPartitionKey                       public String getPk() { return pk; }
    @DynamoDbSortKey                            public String getSk() { return sk; }

    @DynamoDbSecondaryPartitionKey(indexNames = "gsi1")
    public String getGsi1pk() { return gsi1pk; }
    @DynamoDbSecondarySortKey(indexNames = "gsi1")
    public String getGsi1sk() { return gsi1sk; }

    @DynamoDbSecondaryPartitionKey(indexNames = "gsi3")
    public String getGsi3pk() { return gsi3pk; }
    @DynamoDbSecondarySortKey(indexNames = "gsi3")
    public String getGsi3sk() { return gsi3sk; }

    // Optimistic locking: the Enhanced Client turns this into a
    // ConditionExpression on every update. No locks, no lost updates.
    @DynamoDbVersionAttribute                   public Long getVersion() { return version; }

    // remaining getters/setters omitted for brevity
}

package com.acme.orders;

import org.springframework.stereotype.Repository;
import software.amazon.awssdk.enhanced.dynamodb.*;
import software.amazon.awssdk.enhanced.dynamodb.model.*;
import software.amazon.awssdk.services.dynamodb.model.*;

import java.util.List;

@Repository
public class OrderRepository {

    private final DynamoDbTable<OrderItem> table;
    private final DynamoDbIndex<OrderItem> gsi1;
    private final software.amazon.awssdk.services.dynamodb.DynamoDbClient raw;

    OrderRepository(DynamoDbEnhancedClient enhanced,
                    software.amazon.awssdk.services.dynamodb.DynamoDbClient raw) {
        this.table = enhanced.table("orders", TableSchema.fromBean(OrderItem.class));
        this.gsi1 = table.index("gsi1");
        this.raw = raw;
    }

    /** Pattern 1 + 6: the order AND its line items, in ONE round trip. */
    public List<OrderItem> findOrderWithItems(String orderId) {
        return table.query(r -> r
                        .queryConditional(QueryConditional
                                .keyEqualTo(k -> k.partitionValue("ORDER#" + orderId)))
                        // Eventually consistent by default = half the RCU.
                        // Explicit here because it is a decision, not an accident.
                        .consistentRead(false))
                .items().stream().toList();
    }

    /**
     * Pattern 2: a customer's orders, newest first, paginated.
     *
     * The .stream().limit(n) mistake: DynamoDB's Limit bounds ITEMS EXAMINED
     * per request, and a client-side limit still pays for everything fetched.
     * Take ONE page and hand the client the LastEvaluatedKey.
     */
    public Page<OrderItem> findCustomerOrders(String customerId, int pageSize,
                                              java.util.Map<String, AttributeValue> exclusiveStartKey) {
        var request = QueryEnhancedRequest.builder()
                .queryConditional(QueryConditional
                        .keyEqualTo(k -> k.partitionValue("CUST#" + customerId)))
                .scanIndexForward(false)        // newest first
                .limit(pageSize)
                .exclusiveStartKey(exclusiveStartKey)
                .build();
        return gsi1.query(request).iterator().next();   // exactly one page
    }

    /**
     * A state transition that is safe under concurrency, without a lock.
     *
     * The condition enforces the state machine IN THE DATABASE. Two concurrent
     * attempts to pay the same order: exactly one succeeds.
     */
    public void markPaid(String orderId) {
        try {
            raw.updateItem(UpdateItemRequest.builder()
                    .tableName("orders")
                    .key(java.util.Map.of(
                            "pk", AttributeValue.fromS("ORDER#" + orderId),
                            "sk", AttributeValue.fromS("META")))
                    .updateExpression(
                            "SET #st = :paid, paidAt = :now REMOVE gsi3pk, gsi3sk")
                    // REMOVE takes the item out of the sparse work-queue GSI
                    .conditionExpression("#st = :pending")
                    .expressionAttributeNames(java.util.Map.of("#st", "status"))
                    .expressionAttributeValues(java.util.Map.of(
                            ":paid", AttributeValue.fromS("PAID"),
                            ":pending", AttributeValue.fromS("PENDING_PAYMENT"),
                            ":now", AttributeValue.fromS(java.time.Instant.now().toString())))
                    .returnConsumedCapacity(ReturnConsumedCapacity.TOTAL)
                    .build());

        } catch (ConditionalCheckFailedException e) {
            // NOT an error. Either it was already paid (a duplicate, fine) or
            // it is in a state that cannot transition to PAID (a real problem).
            // The caller decides; this is a domain outcome.
            throw new IllegalOrderStateException(orderId, e);
        }
    }
}

package com.acme.catalogue;

import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;

import java.time.Duration;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.ReentrantLock;

@Service
public class CatalogueService {

    private static final Duration BASE_TTL = Duration.ofMinutes(10);
    private static final double JITTER = 0.2;            // plus or minus 20%
    private static final String NEGATIVE_MARKER = "__ABSENT__";

    private final StringRedisTemplate redis;
    private final ProductRepository repo;
    // Per-key locks: request coalescing. On a miss, ONE thread refills and the
    // others wait for its result, instead of 1,000 threads all hitting the DB.
    private final ConcurrentHashMap<String, ReentrantLock> locks = new ConcurrentHashMap<>();

    CatalogueService(StringRedisTemplate redis, ProductRepository repo) {
        this.redis = redis; this.repo = repo;
    }

    public Optional<Product> get(String sku) {
        String key = "catalogue:" + sku;

        try {
            String cached = redis.opsForValue().get(key);
            if (NEGATIVE_MARKER.equals(cached)) return Optional.empty();
            if (cached != null) return Optional.of(deserialize(cached));
        } catch (Exception e) {
            // A cache outage must degrade latency and cost, NEVER availability.
            // Fall through to the database.
            log.warn("cache read failed, falling through", e);
            return repo.findBySku(sku);
        }

        ReentrantLock lock = locks.computeIfAbsent(key, k -> new ReentrantLock());
        lock.lock();
        try {
            // Double-check: another thread may have filled it while we waited.
            String cached = redis.opsForValue().get(key);
            if (NEGATIVE_MARKER.equals(cached)) return Optional.empty();
            if (cached != null) return Optional.of(deserialize(cached));

            Optional<Product> fromDb = repo.findBySku(sku);

            try {
                if (fromDb.isPresent()) {
                    redis.opsForValue().set(key, serialize(fromDb.get()), jitteredTtl());
                } else {
                    // Negative caching with a SHORT ttl: stops a missing-SKU
                    // hammer from hitting the database on every request.
                    redis.opsForValue().set(key, NEGATIVE_MARKER, Duration.ofSeconds(30));
                }
            } catch (Exception ignored) {
                // A cache write failure is not fatal.
            }
            return fromDb;

        } finally {
            lock.unlock();
            locks.remove(key, lock);
        }
    }

    /**
     * TTL jitter. Without it, 2 million keys written during a deploy all expire
     * in the same second, ten minutes later, and the database falls over.
     * This is three lines and it prevents S15.
     */
    private Duration jitteredTtl() {
        long base = BASE_TTL.toSeconds();
        long delta = (long) (base * JITTER * (Math.random() * 2 - 1));
        return Duration.ofSeconds(base + delta);
    }

    /** Write path: invalidate rather than update, so we never cache a stale write. */
    public void update(Product p) {
        repo.save(p);
        try { redis.delete("catalogue:" + p.sku()); } catch (Exception ignored) { }
    }
}

aws dynamodb create-table \
  --table-name orders \
  --billing-mode PAY_PER_REQUEST \
  --attribute-definitions \
      AttributeName=pk,AttributeType=S AttributeName=sk,AttributeType=S \
      AttributeName=gsi1pk,AttributeType=S AttributeName=gsi1sk,AttributeType=S \
  --key-schema AttributeName=pk,KeyType=HASH AttributeName=sk,KeyType=RANGE \
  --global-secondary-indexes '[{
      "IndexName":"gsi1",
      "KeySchema":[{"AttributeName":"gsi1pk","KeyType":"HASH"},
                   {"AttributeName":"gsi1sk","KeyType":"RANGE"}],
      "Projection":{"ProjectionType":"INCLUDE",
                    "NonKeyAttributes":["orderId","status","amount","createdAt"]}
  }]' \
  --stream-specification StreamEnabled=true,StreamViewType=NEW_AND_OLD_IMAGES \
  --sse-specification Enabled=true

aws dynamodb update-continuous-backups --table-name orders \
  --point-in-time-recovery-specification PointInTimeRecoveryEnabled=true

# The idempotency table for Day 8. TTL means it cleans itself up for free.
aws dynamodb create-table --table-name idempotency \
  --billing-mode PAY_PER_REQUEST \
  --attribute-definitions AttributeName=pk,AttributeType=S \
  --key-schema AttributeName=pk,KeyType=HASH
aws dynamodb update-time-to-live --table-name idempotency \
  --time-to-live-specification "Enabled=true,AttributeName=expiresAt"

# (a) ONE Query returns the order and all its items - the single-table payoff
aws dynamodb query --table-name orders \
  --key-condition-expression "pk = :pk" \
  --expression-attribute-values '{":pk":{"S":"ORDER#o-8812"}}' \
  --return-consumed-capacity TOTAL \
  --query '{items: Count, capacity: ConsumedCapacity.CapacityUnits}'
# e.g. {"items": 4, "capacity": 0.5}   <- four items, half an RCU

# (b) THE CONDITIONAL WRITE TEST - the point of the lab.
# Five concurrent attempts to pay the same order. Exactly one must win.
for i in 1 2 3 4 5; do
  curl -s -XPOST "http://localhost:8080/orders/o-8812/pay" -o "/tmp/pay-$i.json" \
       -w "%{http_code}\n" &
done
wait
sort /tmp/pay-*.json | uniq -c
# Expect exactly ONE 200 and four 409-equivalents
# (ConditionalCheckFailedException becomes IllegalOrderStateException)

aws dynamodb get-item --table-name orders \
  --key '{"pk":{"S":"ORDER#o-8812"},"sk":{"S":"META"}}' \
  --query 'Item.{status:status.S,paidAt:paidAt.S}'
# Exactly one paidAt timestamp. The state machine held, with no locks.

# (c) SCAN vs QUERY - the cost lesson, measured
aws dynamodb scan --table-name orders \
  --filter-expression "#s = :s" \
  --expression-attribute-names '{"#s":"status"}' \
  --expression-attribute-values '{":s":{"S":"PENDING_PAYMENT"}}' \
  --return-consumed-capacity TOTAL \
  --query '{scanned: ScannedCount, returned: Count, capacity: ConsumedCapacity.CapacityUnits}'
# {"scanned": 850, "returned": 12, "capacity": 106.5}
# YOU PAID FOR 850 ITEMS TO GET 12.

aws dynamodb query --table-name orders --index-name gsi3 \
  --key-condition-expression "gsi3pk = :p" \
  --expression-attribute-values '{":p":{"S":"PENDING"}}' \
  --return-consumed-capacity TOTAL \
  --query '{scanned: ScannedCount, returned: Count, capacity: ConsumedCapacity.CapacityUnits}'
# {"scanned": 12, "returned": 12, "capacity": 1.5}
# Same answer, about 70x cheaper. THIS is why key design is the whole job.

# (d) Watch the sparse index shrink as work completes
aws dynamodb query --table-name orders --index-name gsi3 \
  --key-condition-expression "gsi3pk = :p" \
  --expression-attribute-values '{":p":{"S":"PENDING"}}' --select COUNT
# Pay some orders, then re-run. The count drops, because items removed the
# GSI3 attributes and therefore left the index. The work queue is self-cleaning.

aws elasticache create-cache-cluster \
  --cache-cluster-id course-cache \
  --engine redis --cache-node-type cache.t4g.micro --num-cache-nodes 1 \
  --cache-subnet-group-name course-private-subnets \
  --security-group-ids "$CACHE_SG" \
  --transit-encryption-enabled
aws elasticache wait cache-cluster-available --cache-cluster-id course-cache

# (a) Cold pass: all misses. Warm pass: hits.
hey -n 2000 -c 20 "http://localhost:8080/catalogue/sku-991"   # first
hey -n 2000 -c 20 "http://localhost:8080/catalogue/sku-991"   # second

# Compare the p99 between the two runs. Expect roughly an order of magnitude.
aws cloudwatch get-metric-statistics --namespace AWS/ElastiCache \
  --metric-name CacheHits --dimensions Name=CacheClusterId,Value=course-cache \
  --start-time "$(date -u -v-15M +%Y-%m-%dT%H:%M:%SZ)" \
  --end-time "$(date -u +%Y-%m-%dT%H:%M:%SZ)" --period 60 --statistics Sum

# (b) CAUSE A STAMPEDE. Warm 1,000 keys in one burst, then expire them all at
# the same instant - exactly what a deploy-time warm does with a fixed TTL.
for i in $(seq 1 1000); do curl -s "http://localhost:8080/catalogue/sku-$i" >/dev/null; done
redis-cli -h "$CACHE_ENDPOINT" --tls --scan --pattern 'catalogue:*' | \
  xargs -n 100 redis-cli -h "$CACHE_ENDPOINT" --tls DEL
# SCAN, not KEYS: Redis is single-threaded and KEYS blocks every other command.

hey -n 5000 -c 500 "http://localhost:8080/catalogue/sku-1"
# Watch the source store: a spike of identical concurrent reads.

# (c) Fix it, then re-measure:
#   1. enable TTL jitter (jitteredTtl())
#   2. enable per-key request coalescing (the ReentrantLock)
# Re-run (b). The source-store read spike should collapse to a handful.

# (d) Prove graceful degradation: a cache outage must not be an outage.
aws elasticache reboot-cache-cluster --cache-cluster-id course-cache \
  --cache-node-ids-to-reboot 0001
# Keep requesting throughout. Expect: higher latency, ZERO errors.
# If you see errors, your cache client is a hard dependency. Fix that.

aws elasticache delete-cache-cluster --cache-cluster-id course-cache
aws dynamodb delete-table --table-name orders
aws dynamodb delete-table --table-name idempotency

50 GB table, 1 KB items            = about 50,000,000 items
4 items fit in one 4 KB read unit
eventually consistent, so 0.5 RCU per unit:
  50,000,000 / 4 x 0.5             = about 6,250,000 read request units per scan
nightly:  6.25M x 30 nights        = about 187M RRU
cost:     187M / 1M x $0.25        = about $47/month

Make the items 4 KB instead of 1 KB, or run it hourly instead of nightly,
and you are quickly into many hundreds of dollars - for 200 rows.

{
  "Effect": "Allow",
  "Action": ["dynamodb:GetItem", "dynamodb:Query"],
  "Resource": "arn:aws:dynamodb:eu-west-1:123456789012:table/orders",
  "Condition": {
    "ForAllValues:StringEquals": {
      "dynamodb:LeadingKeys": ["CUST#${aws:PrincipalTag/customerId}"],
      "dynamodb:Attributes": ["pk","sk","orderId","status","amount"]
    }
  }
}

On-demand:     260M / 1M x $1.25                        = about $325/month
Provisioned:   peak 833 writes/s x 1 KB = 833 WCU
               provision about 1,000 WCU with autoscaling
               1,000 x $0.095                           = about  $95/month

Saving: about $230/month, in exchange for managing autoscaling.
And with a 24-hour TTL, storage stays tiny instead of growing forever.