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

// Diagnostic metric signature:
// 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

import { createHash } from 'crypto'

const SHARD_COUNT = 10 // Spreads writes across 10 distinct physical partition keys

// Deterministic hashing: same orderId always routes to the exact same shard
function getDeterministicShardedKey(orderId: string): string {
  const hash = parseInt(createHash('md5').update(orderId).digest('hex').slice(0, 8), 16)
  const shard = hash % SHARD_COUNT
  return `STATUS#PENDING#${shard}`
}

// Write: distributed evenly across STATUS#PENDING#0 through STATUS#PENDING#9
await docClient.send(new PutCommand({
  TableName: 'AppTable',
  Item: {
    PK: getDeterministicShardedKey(orderId),
    SK: `ORDER#${orderId}`,
    orderId,
    amount: 149.99,
  },
}))

// Direct Point Read: single round-trip without scatter-gather because orderId is known
async function getOrder(orderId: string) {
  return docClient.send(new GetCommand({
    TableName: 'AppTable',
    Key: {
      PK: getDeterministicShardedKey(orderId),
      SK: `ORDER#${orderId}`,
    },
  }))
}

// For workloads where writes never need to be looked up by a single key,
// but are processed in bulk (e.g., streaming audit logs or background worker queues)
function getRandomShardedKey(prefix: string, maxShards: number): string {
  const randomShard = Math.floor(Math.random() * maxShards)
  return `${prefix}#${randomShard}`
}

// Ingestion write: uniform random distribution across 20 shards
await docClient.send(new PutCommand({
  TableName: 'AppTable',
  Item: {
    PK: getRandomShardedKey('EVENT#AUDIT', 20),
    SK: `TS#${Date.now()}#${crypto.randomUUID()}`,
    payload: auditData,
  },
}))

// Bulk Read: scatter-gather across all 20 shards in parallel
async function drainAuditEvents(shardCount: number) {
  const queries = Array.from({ length: shardCount }, (_, i) =>
    docClient.send(new QueryCommand({
      TableName: 'AppTable',
      KeyConditionExpression: 'PK = :pk',
      ExpressionAttributeValues: { ':pk': `EVENT#AUDIT#${i}` },
      Limit: 100,
    }))
  )
  const pages = await Promise.all(queries)
  return pages.flatMap((page) => page.Items ?? [])
}

// When a single logical item (e.g., flash sale SKU or event ticketing inventory)
// experiences momentary write bursts exceeding 1,000 writes/sec,
// application-level sharding fails because all writes target the same physical counter.

// Architecture: Decouple synchronous client writes via SQS FIFO
// Client Request → API Gateway → SQS FIFO (MessageGroupId = skuId) → Lambda Processor → DynamoDB

export const handler = async (event: SQSEvent) => {
  // Lambda receives batches of up to 10 purchases for the same SKU
  // Consolidates 10 individual decrements into a single atomic mutation
  const totalQuantity = event.Records.reduce((sum, r) => sum + JSON.parse(r.body).quantity, 0)
  
  await docClient.send(new UpdateCommand({
    TableName: 'AppTable',
    Key: { PK: 'PRODUCT#taylor-swift-tix', SK: 'INVENTORY' },
    UpdateExpression: 'SET inventory = inventory - :total',
    ConditionExpression: 'inventory >= :total',
    ExpressionAttributeValues: { ':total': totalQuantity },
  }))
}

// For read hotspots (the "Celebrity Profile" problem) where a single item
// is requested 50,000 times/sec, blowing past the 3,000 RCU/sec limit.
// Sharding a single profile item is impossible.

// Solution: Amazon DynamoDB Accelerator (DAX) in-memory cache layer
import { DynamoDBDocumentClient } from '@aws-sdk/lib-dynamodb'
import AmazonDaxClient from 'amazon-dax-client'

// DAX client provides seamless API-compatible microsecond in-memory reads
const daxClient = new AmazonDaxClient({ endpoints: ['dax-cluster.xyz.dax-clusters.cache.amazonaws.com:8111'] })
const docClientWithDax = DynamoDBDocumentClient.from(daxClient)

// Point read: hits in-memory microsecond cache, bypassing DynamoDB storage nodes completely
const celebrityProfile = await docClientWithDax.send(new GetCommand({
  TableName: 'AppTable',
  Key: { PK: 'USER#celebrity', SK: 'PROFILE' },
}))

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,
}))

// ❌ Naive atomic counter — NOT idempotent, unbounded
// On timeout retry, decrements TWICE. If inventory = 0, decrements to -1.
await docClient.send(new UpdateCommand({
  TableName: 'AppTable',
  Key: { PK: 'PRODUCT#p1', SK: 'INVENTORY' },
  UpdateExpression: 'ADD inventory :delta',
  ExpressionAttributeValues: { ':delta': -1 },
}))

// ✅ Bounded atomic decrement — atomic, no prior read required, stock never drops below zero
await docClient.send(new UpdateCommand({
  TableName: 'AppTable',
  Key: { PK: 'PRODUCT#p1', SK: 'INVENTORY' },
  UpdateExpression: 'SET inventory = inventory - :qty',
  ConditionExpression: 'inventory >= :qty',
  ExpressionAttributeValues: { ':qty': 1 },
}))
// If inventory < 1 → ConditionalCheckFailedException (zero state mutation, out of stock)

// Prevents duplicate API requests or retried webhooks from overwriting existing data
await docClient.send(new PutCommand({
  TableName: 'AppTable',
  Item: {
    PK: `PAYMENT#${paymentIntentId}`,
    SK: 'METADATA',
    amount: 99.00,
    status: 'AUTHORIZED',
    createdAt: new Date().toISOString(),
  },
  ConditionExpression: 'attribute_not_exists(PK)', // First write wins; retries fail safely
}))

// Value-matching OCC (e.g., status = :expected) is vulnerable to the ABA problem:
// Thread A reads "ACTIVE". Thread B sets "SUSPENDED". Thread C restores "ACTIVE".
// Thread A's update succeeds despite two intermediate mutations occurring!

// ✅ Monotonic version integer guarantees strict chronological isolation
await docClient.send(new UpdateCommand({
  TableName: 'AppTable',
  Key: { PK: 'ORDER#o99', SK: 'METADATA' },
  UpdateExpression: 'SET deliveryAddress = :addr, version = version + :inc',
  ConditionExpression: 'version = :currentVersion',
  ExpressionAttributeValues: {
    ':addr': '742 Evergreen Terrace',
    ':currentVersion': existingOrder.version,
    ':inc': 1,
  },
}))

// Enforces valid lifecycle transitions directly in the storage engine
// An order can transition to CANCELLED only if currently in PENDING or REVIEW status
await docClient.send(new UpdateCommand({
  TableName: 'AppTable',
  Key: { PK: 'ORDER#o99', SK: 'METADATA' },
  UpdateExpression: 'SET #status = :newStatus, cancelledAt = :now',
  ConditionExpression: 'attribute_exists(PK) AND #status IN (:pending, :review)',
  ExpressionAttributeNames: {
    '#status': 'status', // Reserved keyword protection
  },
  ExpressionAttributeValues: {
    ':newStatus': 'CANCELLED',
    ':pending': 'PENDING',
    ':review': 'REVIEW',
    ':now': new Date().toISOString(),
  },
}))

// Acquire a distributed lock without Redis; auto-expires if worker crashes
const now = Date.now()
const leaseTtl = now + 30_000 // 30-second lease window

await docClient.send(new UpdateCommand({
  TableName: 'AppTable',
  Key: { PK: 'LOCK#reconciliation', SK: 'LOCK' },
  UpdateExpression: 'SET leaseOwner = :workerId, leaseExpiresAt = :expiresAt',
  ConditionExpression: 'attribute_not_exists(leaseOwner) OR leaseExpiresAt < :now',
  ExpressionAttributeValues: {
    ':workerId': 'worker-node-42',
    ':expiresAt': leaseTtl,
    ':now': now,
  },
}))

// 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') }
        }
      }
    })
  ]
}))