DynamoDB 事件存储适应器

德里·卡斯托尔 EventStorageAdapter 使用 AWS 迪纳摩DB。 。 。 .

* 安装

# npm
npm install @castore/event-storage-adapter-dynamodb

# yarn
yarn add @castore/event-storage-adapter-dynamodb

这个软件包 @castore/core 财务报告和审定财务报表 @aws-sdk/client-dynamodb (以上 v3) 作为对等依赖, 您也必须安装它们 :

# npm
npm install @castore/core @aws-sdk/client-dynamodb

# yarn
yarn add @castore/core @aws-sdk/client-dynamodb

目录

这个图书馆暴露出两个适配器:

• DynamoDBSingleTableEventStorageAdapter 它可以将多个事件存储器插到单个的DynamoDB表格上.
• (单位:千美元)已贬值(中文(简体) ) DynamoDBEventStorageAdapter 它需要一个DynamoDB表 每个事件商店.

遗产 DynamoDBEventStorageAdapter 仍然暴露在后向兼容性中。 它会被腐烂并重新命名 LegacyDynamoDBEventStorageAdapter 在v2中,最终去掉在v3中.

联合国

• DynamoDBSingleTableEventStorageAdapter
  • 使用量
  • 如何运作
  • 实例
    • 云阵
    • 光盘
    • 地形
  • 活动组
  • ——————————————
  • 📸 ImageParser
• DynamoDBEventStorageAdapter

DynamoDBSingleTableEventStorageAdapter

使用量

import { DynamoDBClient } from '@aws-sdk/client-dynamodb';

import { DynamoDBSingleTableEventStorageAdapter } from '@castore/event-storage-adapter-dynamodb';

const dynamoDBClient = new DynamoDBClient({});

const pokemonsEventStorageAdapter = new DynamoDBSingleTableEventStorageAdapter(
  {
    tableName: 'my-table-name',
    dynamoDBClient,
  },
);

// 👇 Alternatively, provide a getter
const pokemonsEventStorageAdapter =
  new DynamoDBSingleTableEventStorageAdapter({
    tableName: () => process.env.MY_TABLE_NAME,
    dynamoDBClient,
  });

const pokemonsEventStore = new EventStore({
  ...
  eventStorageAdapter: pokemonsEventStorageAdapter,
});

这将直接将您的事件Store 插入 DynamoDB QQ

如何运作

此适配器在 单独的分区: 当某一事件持续存在时 aggregateId,前缀为 eventStoreId,用作分区键(字符串 属性)及其属性 version 用作排序键( e)编号 属性).

A 类 全球二级指数 也要求有效取回事件存储集合id(listAggregateIds 操作)。 仅是初始事件(version = 1预测数。 A 类 KEYS_ONLY 投影类型已经足够.

// 👇 Initial event
{
  "aggregateId": "POKEMONS#123", // <= Partition key
  "version": 1, // <= Sort key
  "eventStoreId": "POKEMONS", // <= initialEvents index partition key
  "timestamp": "2022-01-01T00:00:00.000Z", // <= initialEvents index sort key
  "type": "POKEMON_APPEARED",
  "payload": { "name": "Pikachu", "level": 42 },
  "metadata": { "trigger": "random" }
}

// 👇 Non-initial event
{
  "aggregateId": "POKEMONS#123",
  "version": 2,
  // Event is not projected on initialEvents index (to limit costs)
  "timestamp": "2023-01-01T00:00:00.000Z",
  "type": "POKEMON_LEVELED_UP"
}

这个 getEvents 方法(用于 getAggregate 财务报告和审定财务报表 getExistingAggregate 工作方法 EventStore 使用一致的读取,同样 始终一致。 。 。 .

这个 pushEvent 方法是一种写操作,所以 始终一致以避免种族条件为条件,如: 卡斯托尔规格。 。 。 .

根据设计 listAggregateIds 操作只能是 最终一致 (通用读取不能一致).

实例

请注意,如果您在 TypeScript 中将您的基础设施定义为代码,您可以直接使用此软件包,而不是硬编码以下数值:

import {
  EVENT_TABLE_PK, // => aggregateId
  EVENT_TABLE_SK, // => version
  EVENT_TABLE_INITIAL_EVENT_INDEX_NAME, // => initialEvents
  EVENT_TABLE_EVENT_STORE_ID_KEY, // => eventStoreId
  EVENT_TABLE_TIMESTAMP_KEY, // => timestamp
} from '@castore/event-storage-adapter-dynamodb';

云阵

{
  "Type": "AWS::DynamoDB::Table",
  "Properties": {
    "AttributeDefinitions": [
      { "AttributeName": "aggregateId", "AttributeType": "S" },
      { "AttributeName": "version", "AttributeType": "N" }
      { "AttributeName": "eventStoreId", "AttributeType": "S" },
      { "AttributeName": "timestamp", "AttributeType": "S" }
    ],
    "KeySchema": [
      { "AttributeName": "aggregateId", "KeyType": "HASH" },
      { "AttributeName": "version", "KeyType": "RANGE" }
    ],
    "GlobalSecondaryIndexes": [
      {
        "IndexName": "initialEvents",
        "KeySchema": [
          { "AttributeName": "eventStoreId", "KeyType": "HASH" },
          { "AttributeName": "timestamp", "KeyType": "RANGE" }
        ],
        "Projection": "KEYS_ONLY"
      }
    ]
  }
}

光盘

import { Table, AttributeType, ProjectionType } from 'aws-cdk-lib/aws-dynamodb';

const { STRING, NUMBER } = AttributeType;
const { KEYS_ONLY } = ProjectionType;

const pokemonsEventsTable = new Table(scope, 'PokemonEvents', {
  partitionKey: {
    name: 'aggregateId',
    type: STRING,
  },
  sortKey: {
    name: 'version',
    type: NUMBER,
  },
});

pokemonsEventsTable.addGlobalSecondaryIndex({
  indexName: 'initialEvents',
  partitionKey: {
    name: 'eventStoreId',
    type: STRING,
  },
  sortKey: {
    name: 'timestamp',
    type: STRING,
  },
  projectionType: KEYS_ONLY,
});

地形

resource "aws_dynamodb_table" "pokemons-events-table" {
  hash_key       = "aggregateId"
  range_key      = "version"

  attribute {
    name = "aggregateId"
    type = "S"
  }

  attribute {
    name = "version"
    type = "N"
  }

  attribute {
    name = "eventStoreId"
    type = "S"
  }

  attribute {
    name = "timestamp"
    type = "S"
  }

  global_secondary_index {
    name               = "initialEvents"
    hash_key           = "eventStoreId"
    range_key          = "timestamp"
    projection_type    = "KEYS_ONLY"
  }
}

活动组

此适配器执行 事件组 使用 DynamoDB 交易 API数字 :

import { EventStore } from '@castore/core';

// 👇 TransactWriteItems N events simultaneously
await EventStore.pushEventGroup(
  // events are correctly typed 🙌
  eventStoreA.groupEvent(eventA1),
  eventStoreA.groupEvent(eventA2),
  eventStoreB.groupEvent(eventB),
  ...
);

注意:

• 交易所涉的所有事件商店都需要使用 DynamoDBSingleTableEventStorageAdapter
• 此工具继承 TransactWriteItem API 密码 限制 : 它可以在同一AWS账户和同一区域的一个或多个DynamoDB表格中针对多达100个不同事件.

——————————————

每次操作所需的 IAM 权限 :

• getEvents (+ 数) getAggregate, (中文(简体) ), getExistingAggregate: dynamodb:Query
• pushEvent数字 : dynamodb:PutItem
• listAggregateIds数字 : dynamodb:Query 编辑 initialEvents 指数

📸 ImageParser

这个图书馆也暴露了一个有用的 ImageParser 要解析的类 迪纳摩DB流 图像来自 a DynamoDBSingleTableEventStorageAdapter。它将构建一个正确的键入 NotificationMessage 流出图像的 uf, 解析它, 删除前缀 aggregateId 并验证 eventStoreId数字 :

import { ImageParser } from '@castore/event-storage-adapter-dynamodb';

const imageParser = new ImageParser({
  sourceEventStores: [pokemonsEventStore, trainersEventStore],
});

// 🙌 Typed as EventStoreNotificationMessage<
//  typeof pokemonsEventStore
//  | typeof trainersEventStore...
// >
const notificationMessage = imageParser.parseImage(
  streamImage,
  // 👇 Optional options
  unmarshallOptions,
);

遗产 DynamoDBEventStorageAdapter

import { DynamoDBClient } from '@aws-sdk/client-dynamodb';

import { DynamoDBEventStorageAdapter } from '@castore/event-storage-adapter-dynamodb';

const dynamoDBClient = new DynamoDBClient({});

const pokemonsEventStorageAdapter = new DynamoDBEventStorageAdapter({
  tableName: 'my-table-name',
  dynamoDBClient,
});

// 👇 Alternatively, provide a getter
const pokemonsEventStorageAdapter = new DynamoDBEventStorageAdapter({
  tableName: () => process.env.MY_TABLE_NAME,
  dynamoDBClient,
});

const pokemonsEventStore = new EventStore({
  ...
  eventStorageAdapter: pokemonsEventStorageAdapter
})

// 👇 Initial event
{
  "aggregateId": "123", // <= Partition key
  "version": 1, // <= Sort key
  "isInitialEvent": 1, // <= initialEvents index partition key
  "timestamp": "2022-01-01T00:00:00.000Z", // <= initialEvents index sort key
  "type": "POKEMON_APPEARED",
  "payload": { "name": "Pikachu", "level": 42 },
  "metadata": { "trigger": "random" }
}

// 👇 Non-initial event
{
  "aggregateId": "123",
  "version": 2,
  // Event is not projected on initialEvents index (to limit costs)
  "timestamp": "2023-01-01T00:00:00.000Z",
  "type": "POKEMON_LEVELED_UP"
}

import {
  EVENT_TABLE_PK, // => aggregateId
  EVENT_TABLE_SK, // => version
  EVENT_TABLE_INITIAL_EVENT_INDEX_NAME, // => initialEvents
  EVENT_TABLE_IS_INITIAL_EVENT_KEY, // => isInitialEvent
  EVENT_TABLE_TIMESTAMP_KEY, // => timestamp
} from '@castore/event-storage-adapter-dynamodb';

{
  "Type": "AWS::DynamoDB::Table",
  "Properties": {
    "AttributeDefinitions": [
      { "AttributeName": "aggregateId", "AttributeType": "S" },
      { "AttributeName": "version", "AttributeType": "N" }
      { "AttributeName": "isInitialEvent", "AttributeType": "N" },
      { "AttributeName": "timestamp", "AttributeType": "S" }
    ],
    "KeySchema": [
      { "AttributeName": "aggregateId", "KeyType": "HASH" },
      { "AttributeName": "version", "KeyType": "RANGE" }
    ],
    "GlobalSecondaryIndexes": [
      {
        "IndexName": "initialEvents",
        "KeySchema": [
          { "AttributeName": "isInitialEvent", "KeyType": "HASH" },
          { "AttributeName": "timestamp", "KeyType": "RANGE" }
        ],
        "Projection": "KEYS_ONLY"
      }
    ]
  }
}

import { Table, AttributeType, ProjectionType } from 'aws-cdk-lib/aws-dynamodb';

const { STRING, NUMBER } = AttributeType;
const { KEYS_ONLY } = ProjectionType;

const pokemonsEventsTable = new Table(scope, 'PokemonEvents', {
  partitionKey: {
    name: 'aggregateId',
    type: STRING,
  },
  sortKey: {
    name: 'version',
    type: NUMBER,
  },
});

pokemonsEventsTable.addGlobalSecondaryIndex({
  indexName: 'initialEvents',
  partitionKey: {
    name: 'isInitialEvent',
    type: NUMBER,
  },
  sortKey: {
    name: 'timestamp',
    type: STRING,
  },
  projectionType: KEYS_ONLY,
});

resource "aws_dynamodb_table" "pokemons-events-table" {
  hash_key       = "aggregateId"
  range_key      = "version"

  attribute {
    name = "aggregateId"
    type = "S"
  }

  attribute {
    name = "version"
    type = "N"
  }

  attribute {
    name = "isInitialEvent"
    type = "N"
  }

  attribute {
    name = "timestamp"
    type = "S"
  }

  global_secondary_index {
    name               = "initialEvents"
    hash_key           = "isInitialEvent"
    range_key          = "timestamp"
    projection_type    = "KEYS_ONLY"
  }
}

import { EventStore } from '@castore/core';

// 👇 TransactWriteItems N events simultaneously
await EventStore.pushEventGroup(
  // events are correctly typed 🙌
  eventStoreA.groupEvent(eventA1),
  eventStoreA.groupEvent(eventA2),
  eventStoreB.groupEvent(eventB),
  ...
);