Table

Source: src/AWS/DynamoDB/Table.ts

An Amazon DynamoDB table with optional indexes, PITR, TTL, and stream-aware binding support.

Table owns the lifecycle of the physical table while the binding contract allows runtime-specific integrations such as Lambda table event sources to request stream configuration without forcing a circular input prop.

Creating Tables

Basic Table

import * as DynamoDB from "alchemy/AWS/DynamoDB";

const table = yield* DynamoDB.Table("UsersTable", {
  partitionKey: "pk",
  attributes: {
    pk: "S",
  },
});

Table with Sort Key and TTL

const table = yield* DynamoDB.Table("SessionsTable", {
  partitionKey: "userId",
  sortKey: "sessionId",
  attributes: {
    userId: "S",
    sessionId: "S",
    expiresAt: "N",
  },
  timeToLiveSpecification: {
    Enabled: true,
    AttributeName: "expiresAt",
  },
});

Table with Global Secondary Index

const table = yield* DynamoDB.Table("OrdersTable", {
  partitionKey: "pk",
  sortKey: "sk",
  attributes: {
    pk: "S",
    sk: "S",
    gsi1pk: "S",
    gsi1sk: "S",
  },
  globalSecondaryIndexes: [{
    indexName: "GSI1",
    partitionKey: "gsi1pk",
    sortKey: "gsi1sk",
    projection: { ProjectionType: "ALL" },
  }],
});

Multi-Attribute GSI Keys

GSI partition and sort keys may be composed of up to four attributes each, indexing natural domain attributes directly instead of synthetic concatenated keys. Partition attributes are hashed together (queries must specify all of them with equality); sort attributes are queried left-to-right in declaration order.

const matches = yield* DynamoDB.Table("TournamentMatches", {
  partitionKey: "matchId",
  attributes: {
    matchId: "S",
    tournamentId: "S",
    region: "S",
    round: "S",
  },
  globalSecondaryIndexes: [{
    indexName: "TournamentRegionIndex",
    partitionKey: ["tournamentId", "region"],
    sortKey: ["round", "matchId"],
    projection: { ProjectionType: "ALL" },
  }],
});

// init
const query = yield* AWS.DynamoDB.Query(matches);

// runtime: query with every partition attribute, then narrow the sort
// attributes left-to-right
const response = yield* query({
  IndexName: "TournamentRegionIndex",
  KeyConditionExpression:
    "tournamentId = :t AND #r = :r AND round = :round",
  ExpressionAttributeNames: { "#r": "region" },
  ExpressionAttributeValues: {
    ":t": { S: "WINTER2024" },
    ":r": { S: "NA-EAST" },
    ":round": { S: "SEMIFINALS" },
  },
});

Runtime Operations

Bind DynamoDB operations in the init phase and use them in runtime handlers. Bindings inject the table name and grant scoped IAM permissions automatically.

// init
const getItem = yield* AWS.DynamoDB.GetItem(table);
const putItem = yield* AWS.DynamoDB.PutItem(table);

return {
  fetch: Effect.gen(function* () {
    // runtime
    yield* putItem({
      Item: { pk: { S: "user#123" }, name: { S: "Alice" } },
    });
    const result = yield* getItem({
      Key: { pk: { S: "user#123" } },
    });
    return yield* HttpServerResponse.json(result.Item);
  }),
};

Table Features

Resource Policy

const table = yield* DynamoDB.Table("SharedTable", {
  partitionKey: "pk",
  attributes: { pk: "S" },
  resourcePolicy: JSON.stringify({
    Version: "2012-10-17",
    Statement: [{
      Effect: "Allow",
      Principal: { AWS: "arn:aws:iam::111122223333:root" },
      Action: ["dynamodb:GetItem", "dynamodb:Query"],
      Resource: "*",
    }],
  }),
});

Kinesis Streaming Destination

import * as Kinesis from "alchemy/AWS/Kinesis";

const stream = yield* Kinesis.Stream("CdcStream", {});
const table = yield* DynamoDB.Table("CdcTable", {
  partitionKey: "pk",
  attributes: { pk: "S" },
  kinesisStreamingDestination: {
    streamArn: stream.streamArn,
    approximateCreationDateTimePrecision: "MICROSECOND",
  },
});

Contributor Insights

const table = yield* DynamoDB.Table("HotKeyTable", {
  partitionKey: "pk",
  attributes: { pk: "S" },
  contributorInsightsEnabled: true,
});

DynamoDB Streams

Process change data capture events from a DynamoDB table using a Lambda event source mapping. The stream is enabled automatically through the binding contract.

// init
yield* DynamoDB.consumeTableChanges(
  table,
  { streamViewType: "NEW_AND_OLD_IMAGES" },
  Effect.fn(function* (record) {
    yield* Effect.log(`${record.eventName}: ${JSON.stringify(record.dynamodb)}`);
  }),
);