記事作成のきっかけ

皆さんはDynamoDBのORMを使っていますか？私は最近DynamoDB自体に触ることがあったのですが、ORMに関する記事がSQLと比べて少なく選択に迷いました。
色々調べた結果、今回のプロジェクトではElectroDBを採用することにしました。（今後の記事でなぜ選んだのか説明する予定です）
これは採用を決めた後に気がついたことですが、公式ドキュメントにはテーブルの作成方法が記載されていません。作成者も「テーブル作成はこのライブラリーの範囲外である」とコメントしており、自分で作成することを考えました。
この記事では自分が採用したテーブル作成方法についてご紹介します。同じ状況で困っている人の助けになれば幸いです。

前提

Typescriptで書いています
SDKのバージョンは@aws-sdk/client-dynamodb": "^3.726.1"です
ElectroDBのバージョンは "electrodb": "^3.0.1",です
BILLING_MODEはPAY_PER_REQUEST固定です
primaryキーのみのEntityは、TypeCheckに引っかかります
アウトプットは、CreateTableCommandInput型です

上記以外のバグや前提条件を発見した場合、コメントでご指摘いただけますと幸いです。

コード

import {
  Entity,
  Service,
  Schema,
  ResponseItem,
  GoQueryTerminalOptions
} from 'electrodb'
import {
  AttributeDefinition,
  GlobalSecondaryIndex,
  KeySchemaElement,
  BillingMode,
  CreateTableCommandInput,
  KeyType,
  ProjectionType,
  ScalarAttributeType
} from '@aws-sdk/client-dynamodb'

import { logger } from '@/lib/logger'

const PRIMARY_INDEX_LABEL = 'PRIMARY' as const
const ELECTRO_INDEX_ATTRIBUTE_TYPE: ScalarAttributeType = 'S' // electrodb uses the composite keys by default
const HASH_KEY_LABEL: KeyType = 'HASH'
const RANGE_KEY_LABEL: KeyType = 'RANGE'
const PROJECTION_TYPE_ALL: ProjectionType = 'ALL'
const PROJECTION_TYPE_KEYS_ONLY: ProjectionType = 'KEYS_ONLY'
const BILLING_MODE_PAY_PER_REQUEST: BillingMode = 'PAY_PER_REQUEST'

type NonSpecificSchema = Schema<string, string, string>
type NonSpecificEntity = Entity<string, string, string, NonSpecificSchema>
type NonSpecificService = Service<Record<string, NonSpecificEntity>>

type MandatoryKeyFields = 'field'
type NonSpecificAccessPattern = NonSpecificSchema['indexes'][string]
type PartitionKey = Pick<NonSpecificAccessPattern['pk'], MandatoryKeyFields>
type SortKey = NonSpecificAccessPattern['sk'] extends undefined
  ? undefined
  : Pick<NonNullable<NonSpecificAccessPattern['sk']>, MandatoryKeyFields>

type ExtractedIndex = Pick<NonSpecificAccessPattern, 'index' | 'project'> & {
  readonly pk: PartitionKey
  readonly sk?: SortKey
}

export type EntityResponseType<E> =
  E extends Entity<infer A, infer F, infer C, infer S>
    ? ResponseItem<A, F, C, S>
    : never

export type EntityQueryOptions<E> = GoQueryTerminalOptions<
  keyof EntityResponseType<E>
>

const generateIndexDefinitions = (
  accessPatterns: NonSpecificAccessPattern[]
): ExtractedIndex[] => {
  const indexMapper = new Map<string, ExtractedIndex>()

  for (const accessPattern of accessPatterns) {
    const indexLabel = accessPattern.index || PRIMARY_INDEX_LABEL
    const existingIndex = indexMapper.get(indexLabel)

    if (!existingIndex) {
      indexMapper.set(indexLabel, {
        index: accessPattern.index,
        project: accessPattern.project,
        pk: {
          field: accessPattern.pk.field
        },
        ...(accessPattern.sk && {
          sk: {
            field: accessPattern.sk.field
          }
        })
      })

      continue
    }

    if (existingIndex.pk.field !== accessPattern.pk.field) {
      throw new Error(
        `Index ${indexLabel} has conflicting fields: '${existingIndex.pk.field}' and '${accessPattern.pk.field}'`
      )
    }

    if (existingIndex.sk?.field !== accessPattern.sk?.field) {
      throw new Error(
        `Index ${indexLabel} has conflicting fields: '${existingIndex.sk?.field}' and '${accessPattern.sk?.field}'`
      )
    }

    // projectは最初の値を優先する
  }

  return Array.from(indexMapper.values())
}

const generateKeySchema = (indexDefinitions: ExtractedIndex[]) => {
  const keySchema: KeySchemaElement[] = []
  const primaryKey = indexDefinitions.find(({ index }) => index === undefined)

  if (!primaryKey) {
    const message = 'Primary index is not defined'
    logger.error(message)
    throw new Error(message)
  }

  const { pk, sk } = primaryKey

  keySchema.push({
    AttributeName: pk.field,
    KeyType: HASH_KEY_LABEL
  })

  if (sk) {
    keySchema.push({
      AttributeName: sk.field,
      KeyType: RANGE_KEY_LABEL
    })
  }

  return keySchema
}

const generateAttributeDefinitions = (indexDefinitions: ExtractedIndex[]) => {
  const attributeDefinitions: AttributeDefinition[] = []

  for (const { pk, sk } of indexDefinitions) {
    attributeDefinitions.push({
      AttributeName: pk.field,
      AttributeType: ELECTRO_INDEX_ATTRIBUTE_TYPE
    })

    if (sk) {
      attributeDefinitions.push({
        AttributeName: sk.field,
        AttributeType: ELECTRO_INDEX_ATTRIBUTE_TYPE
      })
    }
  }

  return attributeDefinitions
}

const generateGlobalSecondaryIndexes = (indexDefinitions: ExtractedIndex[]) => {
  const globalSecondaryIndexes: GlobalSecondaryIndex[] = []

  for (const { pk, sk, index, project } of indexDefinitions) {
    if (index === undefined) continue

    const KeySchema: KeySchemaElement[] = [
      {
        AttributeName: pk.field,
        KeyType: HASH_KEY_LABEL
      }
    ]

    if (sk) {
      KeySchema.push({
        AttributeName: sk.field,
        KeyType: RANGE_KEY_LABEL
      })
    }

    const ProjectionType = project
      ? PROJECTION_TYPE_KEYS_ONLY
      : PROJECTION_TYPE_ALL

    globalSecondaryIndexes.push({
      IndexName: index,
      KeySchema,
      Projection: {
        ProjectionType
      }
    })
  }

  // ProvisionedThroughputはPAY_PER_REQUESTに固定
  return globalSecondaryIndexes
}

const extractEntityIndexes = (entity: NonSpecificEntity) =>
  Object.values(entity.schema.indexes)

export const createSchemaFromService = <S>(
  service: S extends Service<infer E> ? Service<E> : NonSpecificService
): CreateTableCommandInput => {
  const TableName = service.getTableName()
  if (TableName === undefined) {
    const msg = 'TableName is not defined'
    logger.error(msg)
    throw new Error(msg)
  }

  const accessPatterns = Object.values(service.entities)
    .map(extractEntityIndexes)
    .flat()
  const indexDefinitions = generateIndexDefinitions(accessPatterns)

  const AttributeDefinitions = generateAttributeDefinitions(indexDefinitions)
  const KeySchema = generateKeySchema(indexDefinitions)
  const GlobalSecondaryIndexes =
    generateGlobalSecondaryIndexes(indexDefinitions)

  const tableDefinition = {
    TableName,
    AttributeDefinitions,
    KeySchema,
    GlobalSecondaryIndexes,
    BillingMode: BILLING_MODE_PAY_PER_REQUEST
  }

  return tableDefinition
}

解説

概要：DynamoDBテーブルの作成に必要な、PrimaryキーとGSIの情報をServiceから抽出しています。

Serviceに含まれるEntityをループし、各Entityから共通するindex情報を抽出
index情報からkeySchema、AttributeDefinitionを作成
index情報からGlobalSecondaryIndexを作成