@ delta- base/ tablekit

应用级活动为三角洲基地提供工具包。 这个软件包提供高级模式和公用设施,用于建立带有三角洲基础的事件源应用程序.

特征

• 核心类型 - 强烈类型的事件、命令和预测
• 记忆事件存储器 - 完成用于测试的事件存储执行
• 原子多结构附录 - 测试命令,将多个流一起写入
• 命令处理器 - 简单和基于决定者模式的命令处理和重试逻辑
• 投影系统 基于接口的预测,有可插件读取模型存储库
• 内在预测 - 读后一致性同步预测
• 处理错误 - 有类型警卫的全面错误类型
• 工具函数 - 事件和命令创建助手

安装

npm install @delta-base/toolkit
# or
pnpm add @delta-base/toolkit

快速启动

基本事件存储使用

import { InMemoryEventStore, createEvent } from '@delta-base/toolkit';

const eventStore = new InMemoryEventStore();

// Create and append events
const event = createEvent('UserRegistered', {
  userId: 'user-123',
  email: '[email protected]',
  name: 'John Doe'
});

await eventStore.appendToStream('user-123', [event]);

// Read events back
const events = await eventStore.readFromStream('user-123');
console.log(events.events); // [{ type: 'UserRegistered', ... }]

原子多结构附录

InMemoryEventStore 执行 multiStreamAppend 因此测试可以像主机事件商店一样, 进行全能或全能的行为.

const result = await eventStore.multiStreamAppend([
  {
    streamId: 'order-123',
    events: [
      createEvent('OrderCreated', {
        orderId: 'order-123',
        customerId: 'customer-456',
      }),
    ],
    expectedStreamVersion: 'no_stream',
  },
  {
    streamId: 'customer-456',
    events: [
      createEvent('CustomerOrderAdded', {
        customerId: 'customer-456',
        orderId: 'order-123',
      }),
    ],
    expectedStreamVersion: 3,
  },
]);

console.log(result.eventsWritten); // 2
console.log(result.events); // Flat list ordered by globalPosition

如果一个预期的版本错误,则不会改变流. 重复流ID被拒绝.

命令处理决定器模式

import { 
  handleCommandWithDecider,
  createCommand,
  InMemoryEventStore,
  type Decider,
  type ReadEvent
} from '@delta-base/toolkit';

// Define your aggregate state
interface UserState {
  id?: string;
  email?: string;
  isActive: boolean;
}

// Define your events
type UserEvent = 
  | { type: 'UserRegistered'; data: { userId: string; email: string; name: string } }
  | { type: 'EmailUpdated'; data: { userId: string; newEmail: string } };

// Define your commands  
type UserCommand = 
  | { type: 'RegisterUser'; data: { userId: string; email: string; name: string } }
  | { type: 'UpdateEmail'; data: { userId: string; newEmail: string } };

// Create your decider
const userDecider: Decider<UserState, UserCommand, UserEvent> = {
  initialState: () => ({ isActive: false }),
  
  decide: (state: UserState, command: UserCommand) => {
    switch (command.type) {
      case 'RegisterUser':
        if (state.id) throw new Error('User already exists');
        return [{
          type: 'UserRegistered',
          data: command.data
        }];
      
      case 'UpdateEmail':
        if (!state.id) throw new Error('User does not exist');
        return [{
          type: 'EmailUpdated', 
          data: { userId: command.data.userId, newEmail: command.data.newEmail }
        }];
      
      default:
        return [];
    }
  },
  
  evolve: (state: UserState, event: ReadEvent<UserEvent>) => {
    switch (event.type) {
      case 'UserRegistered':
        return {
          ...state,
          id: event.data.userId,
          email: event.data.email,
          isActive: true
        };
      
      case 'EmailUpdated':
        return {
          ...state,
          email: event.data.newEmail
        };
      
      default:
        return state;
    }
  }
};

// Use the command handler
const eventStore = new InMemoryEventStore();

const command = createCommand('RegisterUser', {
  userId: '123',
  email: '[email protected]', 
  name: 'John Doe'
});

const result = await handleCommandWithDecider(
  eventStore,
  'user-123',
  command,
  userDecider
);

console.log(result.newState); // { id: '123', email: '[email protected]', isActive: true }
console.log(result.newEvents); // [UserRegistered event]

用新接口系统构建预测

新的投影系统提供了一个强大的基于接口的方法,有可插接的读取模型存储:

import { 
  BaseProjection,
  InMemoryReadModelStore,
  KVReadModelStore,
  type ReadEvent,
  type IReadModelStore
} from '@delta-base/toolkit';

// Define your read model structure
interface UserReadModel {
  id: string;
  email: string;
  name: string;
  status: 'active' | 'inactive';
  registeredAt: string;
  lastUpdated: string;
  revision: number;
}

// Create a projection class extending BaseProjection
class UserProjection extends BaseProjection {
  readonly supportedEventTypes = [
    'UserRegisteredEvent',
    'UserUpdatedEvent', 
    'UserStatusChangedEvent'
  ];

  constructor(store: IReadModelStore) {
    super(store);
  }

  protected async processEvent(event: ReadEvent): Promise<void> {
    switch (event.type) {
      case 'UserRegisteredEvent':
        await this.handleUserRegistered(event);
        break;
      case 'UserUpdatedEvent':
        await this.handleUserUpdated(event);
        break;
      case 'UserStatusChangedEvent':
        await this.handleUserStatusChanged(event);
        break;
    }
  }

  private async handleUserRegistered(event: ReadEvent): Promise<void> {
    const eventData = event.data as any;
    const userId = `user:${eventData.userId}`;
    
    // Check if user already exists (idempotency)
    const existingUser = await this.store.get<UserReadModel>(userId);
    if (existingUser && !(await this.shouldProcessEvent(event, existingUser.revision))) {
      return;
    }

    const userReadModel: UserReadModel = {
      id: eventData.userId,
      email: eventData.email,
      name: eventData.name,
      status: 'active',
      registeredAt: event.createdAt,
      lastUpdated: event.createdAt,
      revision: event.streamPosition
    };

    await this.store.put(userId, userReadModel);
  }

  private async handleUserUpdated(event: ReadEvent): Promise<void> {
    const eventData = event.data as any;
    const userId = `user:${eventData.userId}`;
    
    const existingUser = await this.store.get<UserReadModel>(userId);
    if (!existingUser) {
      console.warn(`User ${eventData.userId} not found for update`);
      return;
    }

    if (!this.validateRevision(event, existingUser.revision)) {
      return;
    }

    const updatedUser: UserReadModel = {
      ...existingUser,
      email: eventData.email || existingUser.email,
      name: eventData.name || existingUser.name,
      lastUpdated: event.createdAt,
      revision: event.streamPosition
    };

    await this.store.put(userId, updatedUser);
  }

  private async handleUserStatusChanged(event: ReadEvent): Promise<void> {
    const eventData = event.data as any;
    const userId = `user:${eventData.userId}`;
    
    const existingUser = await this.store.get<UserReadModel>(userId);
    if (!existingUser || !this.validateRevision(event, existingUser.revision)) {
      return;
    }

    const updatedUser: UserReadModel = {
      ...existingUser,
      status: eventData.status,
      lastUpdated: event.createdAt,
      revision: event.streamPosition
    };

    await this.store.put(userId, updatedUser);
  }
}

// Usage with different store implementations
const inMemoryStore = new InMemoryReadModelStore();
const userProjection = new UserProjection(inMemoryStore);

// Process events
await userProjection.processEvents([userRegisteredEvent, userUpdatedEvent]);

// Query the read model
const user = await inMemoryStore.get<UserReadModel>('user:123');
console.log(user); // { id: '123', email: '[email protected]', ... }

读取模型存储执行

该工具包提供多个存储执行:

内存存储(用于测试)

import { InMemoryReadModelStore } from '@delta-base/toolkit';

const store = new InMemoryReadModelStore();

// Supports all IReadModelStore operations
await store.put('key', { data: 'value' });
const value = await store.get('key');
const allItems = await store.getAll({ prefix: 'user:' });

// Advanced features
await store.batchPut([
  { key: 'user:1', value: { name: 'John' } },
  { key: 'user:2', value: { name: 'Jane' } }
]);

const users = await store.query({
  filter: { status: 'active' }
});

Cloudflare KV 存储

import { KVReadModelStore } from '@delta-base/toolkit';

// In a Cloudflare Worker
const store = new KVReadModelStore(env.MY_KV_NAMESPACE);

// Supports native batch operations for performance
const users = await store.batchGet(['user:1', 'user:2', 'user:3']);

// TTL and metadata support
await store.put('session:123', sessionData, {
  expirationTtl: 3600, // 1 hour
  metadata: { userId: '123' }
});

HTTP 存储器( 用于外部 API)

import { HttpReadModelStore } from '@delta-base/toolkit';

const store = new HttpReadModelStore('https://api.example.com/readmodels', {
  'Authorization': 'Bearer token'
});

// Works with any HTTP API that follows REST conventions
await store.put('user:123', userData);
const user = await store.get('user:123');

内在预测(同步)

线性预测与事件附件相同,提供读后写的一致性。 公开界定预测 singleStreamProjection 或者说 multiStreamProjection。 。 。 .

单层投影

每行读取一个模型文档. 该文件 _version 跟踪流位置,以进行一能和追赶.

import {
  singleStreamProjection,
  projections,
  InMemoryEventStore,
  InMemoryReadModelStore,
} from '@delta-base/toolkit';

type OrderSummary = {
  items: Array<{ name: string; price: number }>;
  totalAmount: number;
  status: string;
};

type OrderEvent =
  | { type: 'order.created'; data: { customerId: string } }
  | { type: 'order.item-added'; data: { item: { name: string; price: number } } }
  | { type: 'order.confirmed'; data: { confirmedAt: string } };

const readModelStore = new InMemoryReadModelStore();

const orderSummary = singleStreamProjection<OrderSummary, OrderEvent>({
  projectionName: 'order-summary',
  store: readModelStore,
  canHandle: ['order.created', 'order.item-added', 'order.confirmed'],
  getDocumentId: (streamId) => `order:${streamId}`,
  initialState: () => ({ items: [], totalAmount: 0, status: 'draft' }),
  evolve: (doc, event) => {
    switch (event.type) {
      case 'order.created':
        return { ...doc, status: 'created' };
      case 'order.item-added':
        return {
          ...doc,
          items: [...doc.items, event.data.item],
          totalAmount: doc.totalAmount + event.data.item.price,
        };
      case 'order.confirmed':
        return { ...doc, status: 'confirmed' };
      default:
        return doc;
    }
  },
});

// Register on event store — projections run on every appendToStream
const eventStore = new InMemoryEventStore({
  projections: projections.inline([orderSummary]),
});

await eventStore.appendToStream('order-1', [
  { type: 'order.created', data: { customerId: 'cust-1' } },
  { type: 'order.item-added', data: { item: { name: 'Widget', price: 9.99 } } },
]);

// Read model is already up-to-date
const order = await readModelStore.get('order:order-1');
// { items: [{ name: 'Widget', price: 9.99 }], totalAmount: 9.99, status: 'created', _version: 2 }

多层投影

将多个流的事件汇总到单个文档中。 用途 globalPosition 用于光标跟踪。 也用于 DCB( 动态一致性边界) 中 getDocumentId 从事件标记中获取密钥.

import { multiStreamProjection } from '@delta-base/toolkit';

type CustomerDashboard = {
  totalOrders: number;
  totalSpent: number;
  lastOrderAt: string | null;
};

const customerDashboard = multiStreamProjection<CustomerDashboard, OrderEvent>({
  projectionName: 'customer-dashboard',
  store: readModelStore,
  canHandle: ['order.created', 'order.confirmed'],
  getDocumentId: (event) => `customer:${event.data.customerId}`,
  initialState: () => ({ totalOrders: 0, totalSpent: 0, lastOrderAt: null }),
  evolve: (doc, event) => {
    switch (event.type) {
      case 'order.created':
        return { ...doc, totalOrders: doc.totalOrders + 1, lastOrderAt: event.createdAt };
      default:
        return doc;
    }
  },
});

混合宣布和强制性预测

这个 projections.inline() 帮助者接受申报预测(来自上述工厂)和必要 Projection<TEvent> 实例 :

const eventStore = new InMemoryEventStore({
  projections: projections.inline([
    orderSummary,                      // declarative
    new OrderAuditProjection(store),   // imperative (implements Projection<T>)
  ]),
});

抓起来

当一个预测在事件已经存在后被记录时,它会赶上 在第一个附件进行现场处理之前自动。 如果以后有预测 失败, 只有投影被标记为停滞 并再次赶上它之前 处理更新的附加事件。 您也可以明确触发追赶:

await eventStore.catchUpProjections();

处理错误

projections.inline([orderSummary], {
  onError: 'throw',  // default: projection failure fails the append call
  // onError: 'log', // log error, append succeeds
  // onError: (err, projection, events) => { ... } // custom handler
});

活动是 从不退后 关于投影失败。 它们是事实. 与 onError: 'throw',附件拒绝在事件持续之后。 与 onError: 'log' 或自定义的非掷出处理器,附件仍然可以解析. 如果追赶失败, DeltaBase 跳过 handleAppend 对于那个陈旧的预测 因此,它不能推进过去错过的历史事件.

目 录

内线预测被设计为一能. 这个 _version 字段跟踪每个文档上一个处理过的流位置(单流)或全局位置(多流)。 处理同一事件两次产生相同的结果. 这在同时存在的环境中至关重要(例如,多个Cloudflare工人隔离). 让你 evolve 函数确定 - 相同的输入总是产生相同的输出.

Webhook 预测

import { 
  createWebhookProjectionHandler,
  KVReadModelStore
} from '@delta-base/toolkit';

// Create projection for webhook deployment
class UserStatsProjection extends BaseProjection {
  readonly supportedEventTypes = ['UserRegisteredEvent', 'UserStatusChangedEvent'];

  protected async processEvent(event: ReadEvent): Promise<void> {
    const stats = await this.store.get<{ totalUsers: number; activeUsers: number }>('stats') || 
                  { totalUsers: 0, activeUsers: 0 };

    switch (event.type) {
      case 'UserRegisteredEvent':
        stats.totalUsers++;
        stats.activeUsers++;
        break;
      case 'UserStatusChangedEvent':
        const eventData = event.data as any;
        if (eventData.status === 'inactive') stats.activeUsers--;
        else if (eventData.status === 'active') stats.activeUsers++;
        break;
    }

    await this.store.put('stats', stats);
  }
}

// In Cloudflare Worker
const projection = new UserStatsProjection(new KVReadModelStore(env.KV_NAMESPACE));
const webhookHandler = createWebhookProjectionHandler(projection);

export default {
  async fetch(request: Request): Promise<Response> {
    return await webhookHandler(request);
  }
};

存储能力和运行时间检测

// Check store capabilities at runtime
const capabilities = store.getCapabilities();

if (capabilities.features.ttl) {
  // Store supports TTL
  await store.put('temp-data', data, { expirationTtl: 300 });
}

if (capabilities.features.advancedQueries) {
  // Store supports complex queries
  const results = await store.query({
    filter: { 
      status: 'active',
      lastLogin: { $gte: new Date('2024-01-01') }
    }
  });
}

console.log(`Store type: ${capabilities.storeType}`);
console.log(`Max batch size: ${capabilities.limits.maxBatchSize}`);

核心概念

活动

事件代表了过去发生的事情。 它们是不可改变的,包含:

• eventId - 独特的标识符(由事件存储所添加)
• type - 事件类型(如“用户注册”)
• streamId - 事件所属的流程(由事件商店添加)
• streamPosition - 流中的位置(由事件存储所添加)
• globalPosition - 全球在所有流中的位置(由活动商店添加)
• data - 事件有效载荷
• metadata - 其他元数据(可选)
• schemaVersion - 事件计划版本(由事件商店添加)
• transactionId - 交易标识符(由事件存储所添加)
• createdAt - 事件发生时(由事件存储所添加)

图标

命令代表着做某事的意图. 其中包括:

• commandId - 独特的标识符(自动添加)
• type - 命令类型(例如“注册用户”)
• data - 指挥有效载荷
• metadata - 其他元数据(可选)
• createdAt - 当命令创建时(自动添加)

决定模式

决定者模式分开:

1. 决定 - 鉴于目前的状况和命令,应该产生什么事件?
2. 演变 - 鉴于目前的状况和事件,什么是新状态?
3. 初始状态 - 什么是开始状态?

这种模式使商业逻辑变得纯正可考.

读取模型预测

新的投影系统提供了一个基于接口的清洁方法:

IReadModelStore 接口

• 统一 API: 相同的接口与内存、 KV、 HTTP 和其他仓库相同
• 灵活操作: 得到、放出、删除、获取全部、批出、批出、查询
• 存储能力: 运行时特性检测以达到最佳性能
• 多表支助: 可选表格/名称空间隔离

投影接口

• 事件过滤: 宣布支持的事件类型用于自动过滤
• 批处理: 高效处理多起事件,以便
• 类型安全: 严格输入事件处理

基准预测类

• 常见模式: 内设审校跟踪和审校能力检查
• 事件顺序: 顺序处理保持一致性
• 处理错误:慷慨地处理不合秩序的事件

福利

• 开发者经验: 具有综合类型脚本支持的清洁 API
• 业绩: 本地批量操作和储存优化查询
• 可缩放性: 从内存到分布式系统的可插件存储
• 测试: 易于用内存商店进行测试
• 生产: 部署到任何具有适当存储执行的平台

处理错误

该工具包提供了全面的有类型卫士的错误类型:

import { 
  isDeltaBaseError, 
  isVersionConflictError,
  StreamVersionConflictError 
} from '@delta-base/toolkit';

try {
  await handleCommand(eventStore, streamId, command, decider);
} catch (error) {
  if (isVersionConflictError(error)) {
    // Handle concurrency conflict
    console.log('Version conflict, retrying...');
  } else if (isDeltaBaseError(error)) {
    // Handle other toolkit errors
    console.log('Toolkit error:', error.message);
  } else {
    // Handle unexpected errors
    console.log('Unexpected error:', error);
  }
}

API 参考

核心类型

• Event<TType, TData, TMetadata?> - 事件界面
• ReadEvent<TEvent> - 从带有系统字段的事件存储处读取的事件
• Command<TType, TData, TMetadata?> - 命令接口
• StreamId - 流式标识符类型
• EventStore - 事件持续界面
• MultiStreamAppendRequest - 原子多流子内的一个流子附件
• MultiStreamAppendResult - 原子多流附件的结果

命令处理

• handleCommand() - 处理命令,不带决定者模式
• handleCommandWithDecider() - 用决定者模式处理命令
• handleCommandWithRetry() - 自动重试处理命令
• handleCommandWithDeciderAndRetry() - 与决定者处理命令并重试
• Decider<State, Command, Event> - 决定模式接口

读取模型预测

• IReadModelStore - 读取模型存储的统一接口
• InMemoryReadModelStore - 模拟试验
• KVReadModelStore - 使用本地批量操作执行Cloudflare KV
• HttpReadModelStore - 外部服务的HTTP API实施
• Projection - 事件驱动预测接口
• BaseProjection - 具有共同预测模式的抽象基类
• createWebhookProjectionHandler() - 创建 HTTP Webhook 处理器

内在预测

• singleStreamProjection(options) - 创建单流内线投影(每流一个文件)
• multiStreamProjection(options) - 创建多流内线投影(跨流聚合)
• projections.inline(items, config?) - 登记内在执行的预测
• InlineProjection<TEvent> - 内含预测与赶超支持的接口
• InlineProjectionRegistration - 登记对象 getEventStore() 选项

数据库

• InMemoryEventStore - 用于测试的全事件存储执行
  • 全部执行 EventStore 接口方法
  • 包括诸如 getAllStreamIds(), (中文(简体) ), clear()等 (中文(简体) ).

错误类型

所有错误都继承自 DeltaBaseError 包括下列具体类型:

• VersionConflictError 编号 : StreamVersionConflictError - 货币冲突
• ValidationError - 请求验证失败
• AuthenticationError 编号 : AuthorizationError - 认证失败
• NotFoundError 编号 : StreamNotFoundError - 未找到资源
• TimeoutError 编号 : RateLimitError - 业务失败

公用事业

• createEvent(type, data, metadata?) - 用类型推论创建事件
• createCommand(type, data, metadata?) - 用类型推论创建命令
• createReadEvent() - 创建测试读取事件

测试

该工具包旨在便于测试:

import { 
  InMemoryEventStore,
  InMemoryReadModelStore,
  createEvent, 
  createCommand,
  handleCommandWithDecider,
  BaseProjection,
  type ReadEvent
} from '@delta-base/toolkit';

describe('User Registration with Projections', () => {
  let eventStore: InMemoryEventStore;
  let readModelStore: InMemoryReadModelStore;
  let userProjection: UserProjection;
  
  beforeEach(() => {
    eventStore = new InMemoryEventStore();
    readModelStore = new InMemoryReadModelStore();
    userProjection = new UserProjection(readModelStore);
  });
  
  it('should register user and update read model', async () => {
    // Given - command
    const command = createCommand('RegisterUser', {
      userId: '123',
      email: '[email protected]',
      name: 'John Doe'
    });
    
    // When - command is handled
    const result = await handleCommandWithDecider(
      eventStore,
      'user-123',
      command,
      userDecider
    );
    
    // And - projection processes the events
    await userProjection.processEvents(result.newEvents);
    
    // Then - verify command result
    expect(result.newState.id).toBe('123');
    expect(result.newEvents).toHaveLength(1);
    expect(result.newEvents[0].type).toBe('UserRegistered');
    
    // And - verify read model
    const userReadModel = await readModelStore.get('user:123');
    expect(userReadModel).toEqual({
      id: '123',
      email: '[email protected]',
      name: 'John Doe',
      status: 'active',
      registeredAt: expect.any(String),
      lastUpdated: expect.any(String),
      revision: 1
    });
  });
});

实例

见 /examples 完整工作示例目录 :

• 基本用户管理 - 附有命令和预测的完整用户生命周期

最佳做法

1. 保持商业逻辑纯度: 使用决定者模式来区分决定和效果
2. A. 设计是否充分: 命令和预测应可安全重试
3. 版本您的事件: 在事件元数据中包含计划版本
4. 用事件测试: 写入验证事件流并读取模型的测试
5. 处理错误: 使用提供的错误类型和类型保护
6. 选择右侧存储: 使用InMemory进行测试, KV 用于无服务器, HTTP 用于现有API
7. 批量操作: 使用批次Get/批次Put来改进性能
8. 监视器存储能力: 在运行时检查能力以达到最佳行为