Hash function: partition_key_value → partition_node

"USER#u001" → partition_node_A
"USER#u002" → partition_node_B
"USER#u003" → partition_node_C
"STATUS#PENDING" → partition_node_A  ← ALL pending orders hit the SAME node

// CloudWatch Insights query: detect hot partition by divergence between consumed and provisioned
fields @timestamp, @message
| filter @message like "ProvisionedThroughputExceededException"
| stats count(*) as throttleCount by bin(5m)
| sort throttleCount desc

// Also monitor these metrics per-table:
// 1. SuccessfulRequestLatency P99 — spikes when partition nodes are stressed
// 2. ThrottledRequests — non-zero = throttling occurring
// 3. ConsumedWriteCapacityUnits vs ProvisionedWriteCapacityUnits divergence
//    → consumed << provisioned but still throttling = hot partition

const SHARD_COUNT = 10 // Number of write shards

// ❌ Hot partition: all PENDING orders → one partition
const hotItem = { PK: 'STATUS#PENDING', SK: orderId, ...orderData }

// ✅ Salted partition key: 10 shards spread writes across 10 partitions
function getShardedKey(orderId: string): string {
  // Deterministic shard assignment: same orderId always gets same shard
  const hash = parseInt(createHash('md5').update(orderId).digest('hex').slice(0, 8), 16)
  const shard = hash % SHARD_COUNT
  return `STATUS#PENDING#${shard}`
}

const shardedItem = {
  PK: getShardedKey(orderId), // e.g., "STATUS#PENDING#3"
  SK: orderId,
  ...orderData
}

// Reading all pending orders: scatter-gather across all shards
async function getAllPendingOrders() {
  const shardQueries = Array.from({ length: SHARD_COUNT }, (_, i) =>
    docClient.send(new QueryCommand({
      TableName: 'AppTable',
      KeyConditionExpression: 'PK = :pk',
      ExpressionAttributeValues: { ':pk': `STATUS#PENDING#${i}` }
    }))
  )
  const results = await Promise.all(shardQueries)
  return results.flatMap(r => r.Items ?? [])
}

Scenario: AppTable writes 1,000 WPS. GSI1 provisioned at 10 WCU.

Every base table write that updates an item with GSI1 key attributes
  → triggers a GSI1 write
  → GSI1 is provisioned at 10 WCU → can only handle 10 writes/second
  → at 1,000 WPS, GSI1 is overwhelmed

Effect: DynamoDB throttles the BASE TABLE write
  → ProvisionedThroughputExceededException on PutItem/UpdateItem calls
  → Error surfaces in the application as a base table write failure
  → CloudWatch shows base table ThrottledWrites = high
  → Engineers increase BASE TABLE WCU → no improvement (GSI is the bottleneck)
  → Root cause: GSI1 capacity, not base table capacity

// Required CloudWatch alarms — one per GSI, not just per table
new Alarm(this, 'GSI1WriteThrottle', {
  metric: new Metric({
    namespace: 'AWS/DynamoDB',
    metricName: 'WriteThrottleEvents',
    dimensionsMap: {
      TableName: 'AppTable',
      GlobalSecondaryIndexName: 'GSI1', // ← Must specify the GSI name
    },
    statistic: 'Sum',
    period: Duration.minutes(1),
  }),
  threshold: 0,
  evaluationPeriods: 1,
  comparisonOperator: ComparisonOperator.GREATER_THAN_THRESHOLD,
  alarmDescription: 'GSI1 write throttling is causing base table write failures',
})

// GSI capacity must be provisioned at the same rate as the base table
// For On-Demand mode: GSI capacity scales with the table automatically
// For Provisioned mode: set GSI RCU/WCU to at least the expected write rate per item

// Use case: order placement — must atomically:
// 1. Create the order item
// 2. Decrement product inventory
// 3. Create the ledger entry
// All three succeed, or none of them succeed

await docClient.send(new TransactWriteCommand({
  TransactItems: [
    {
      Put: {
        TableName: 'AppTable',
        Item: {
          PK: `ORDER#${orderId}`,
          SK: 'METADATA',
          status: 'PENDING',
          total: 149.99,
          userId,
        },
        ConditionExpression: 'attribute_not_exists(PK)', // Idempotency guard
      }
    },
    {
      Update: {
        TableName: 'AppTable',
        Key: { PK: `PRODUCT#${productId}`, SK: 'INVENTORY' },
        UpdateExpression: 'SET inventory = inventory - :qty',
        ConditionExpression: 'inventory >= :qty', // Prevent negative inventory
        ExpressionAttributeValues: { ':qty': quantity }
      }
    },
    {
      Put: {
        TableName: 'LedgerTable', // Cross-table transaction
        Item: {
          PK: `LEDGER#${orderId}`,
          SK: `DEBIT#${Date.now()}`,
          amount: -149.99,
          reason: 'order_payment',
        }
      }
    }
  ],
  // Idempotency token: safe to retry with same token — duplicate has no effect
  ClientRequestToken: idempotencyKey,
}))

// ❌ Atomic counter — NOT idempotent
// On a retry, the counter increments TWICE for one logical operation
await docClient.send(new UpdateCommand({
  TableName: 'AppTable',
  Key: { PK: 'PRODUCT#p1', SK: 'INVENTORY' },
  UpdateExpression: 'ADD inventory :delta',
  ExpressionAttributeValues: { ':delta': -1 }
  // Problem: on network timeout, the caller retries → ADD -1 fires TWICE
  // Inventory decremented by 2 for one purchase
}))

// ✅ Conditional write — idempotent optimistic concurrency
// If inventory has already changed (another write snuck in), the condition fails safely
await docClient.send(new UpdateCommand({
  TableName: 'AppTable',
  Key: { PK: 'PRODUCT#p1', SK: 'INVENTORY' },
  UpdateExpression: 'SET inventory = :newValue',
  ConditionExpression: 'inventory = :expectedValue',
  ExpressionAttributeValues: {
    ':expectedValue': currentInventory,  // The value we read
    ':newValue': currentInventory - 1,   // What we want it to be
  }
  // On retry: if the write already succeeded, inventory = currentInventory - 1 ≠ currentInventory
  // → ConditionalCheckFailedException → treat as success (the intended write happened)
  // On concurrent write: another thread changed inventory → ConditionalCheckFailedException
  // → treat as conflict → re-read and retry with fresh expectedValue
}))

// CDK: DynamoDB Streams with Lambda ESM and filter expressions
import { Table, StreamViewType } from 'aws-cdk-lib/aws-dynamodb'
import { DynamoEventSource, StartingPosition } from 'aws-cdk-lib/aws-lambda-event-sources'
import { FilterCriteria, FilterRule } from 'aws-cdk-lib/aws-lambda'

const table = new Table(this, 'AppTable', {
  stream: StreamViewType.NEW_AND_OLD_IMAGES, // Best for audit; most expensive (full payloads)
})

// ESM filter: only invoke Lambda for INSERT events on ORDER items
// Eliminates unnecessary Lambda invocations for other entity types
processorFn.addEventSource(new DynamoEventSource(table, {
  startingPosition: StartingPosition.TRIM_HORIZON,
  batchSize: 100,
  bisectBatchOnFunctionError: true,
  maxRecordAge: Duration.hours(1),
  retryAttempts: 3,
  onFailure: new SqsDestination(dlq),
  filters: [
    FilterCriteria.filter({
      eventName: FilterRule.isEqual('INSERT'),
      dynamodb: {
        NewImage: {
          entityType: { S: FilterRule.isEqual('ORDER') }
        }
      }
    })
  ]
}))