# @crossdelta/cloudevents 中文文档

规范地址：https://npmcn.com/package/%40crossdelta/cloudevents/

版本：0\.10\.1
npm 元数据更新时间：2026-10-25T12:43:36.372Z
中文译文更新时间：2026-11-11T01:35:42.312Z
软件包许可：MIT

本文由上游 README 翻译，可能存在偏差，请结合原文核对 API、命令与配置。版权与许可归原作者所有。

原始 npm 页面：https://www.npmjs.com/package/%40crossdelta/cloudevents
原文对照：https://npmcn.com/package/%40crossdelta/cloudevents/?lang=original

## 关键信息（来自 npm 元数据）

- 包体积（解压后）：408 kB，7 个文件
- 模块格式：ESM + CJS
- TypeScript 类型：包内自带
- 运行依赖：6 个
- peer 依赖：zod ^4\.0\.0，hono ^4\.12\.12，@faker-js/faker \>=9\.0\.0，@google-cloud/pubsub \>=4\.0\.0，@crossdelta/telemetry ^0\.13\.2
- 首次发布：2025-12-02T18:31:43.313Z
- 最新版本发布：2026-10-25T12:43:36.181Z
- 近 30 天下载量：4041（2026-10-11 至 2026-11-09，npm 官方下载统计）
- 已发布版本数：71（近 12 个月 71 个）

## 中文简介

用于类型Script的 CloudEvents 工具包 - Zod验证,处理器发现,NATS JetStream & Core

## 安装

```sh
npm install @crossdelta/cloudevents
```

也可以使用：`pnpm add @crossdelta/cloudevents`、`yarn add @crossdelta/cloudevents`、`bun add @crossdelta/cloudevents`

## README

# @ crossdelta/ clodevents (英语)

类型安全事件驱动微服务 [海军](https://nats.io/) 财务报告和审定财务报表 [佐德语Name](https://zod.dev/) 验证,使用 [云节](https://cloudevents.io/) 规格.

```
                         NATS JetStream
                        ┌──────────────┐
  ┌──────────────┐      │              │      ┌──────────────┐
  │    Service   │      │   Stream:    │      │   Service    │
  │   (publish)  │─────▶│   ORDERS     │─────▶│  (consume)   │
  └──────────────┘      │              │      └──────────────┘
                        └──────────────┘
        │                                            │
        │  publish(...)                              │  handleEvent(...)
        ▼                                            ▼
  ┌──────────────┐                            ┌──────────────┐
  │ { orderId,   │                            │ Zod schema   │
  │   total }    │                            │ validates    │
  └──────────────┘                            └──────────────┘
```

```bash
bun add @crossdelta/cloudevents zod@4
```

> **先决条件:** 运行 [NATS 服务器](https://docs.nats.io/running-a-nats-service/introduction) 已启用 JetStream .
> 
> **说明:** 需要 Zod v4 支持全部类型脚本 .

## 快速启动

> **说明:** 服务 **从不创建流**在服务消费前,必须先有流:
> 
> -   **发展**: 使用 [内兹克州](https://docs.nats.io/running-a-nats-service/nats_admin/jetstream_admin/streams) 或者说 [平台 SDK](https://www.npmjs.com/package/@crossdelta/platform-sdk) (单位:千美元)`pf dev` 自动创建合同)
> -   **生产**: 使用基础设施即用代码(Pulumi、Terraform)或 [串流设置](https://npmcn.com/package/%40crossdelta/cloudevents/#readme-stream-setup) 下面的函数

**1\. 创建事件处理器** (单位:千美元)`src/events/order-created.handler.ts`:

```typescript
import { handleEvent } from '@crossdelta/cloudevents'
import { z } from 'zod'

// Export schema for mock generation
export const OrderCreatedSchema = z.object({
  orderId: z.string(),
  total: z.number(),
})

// Export type for use in use-cases
export type OrderCreatedEvent = z.infer<typeof OrderCreatedSchema>

export default handleEvent(
  {
    schema: OrderCreatedSchema,
    type: 'order.created',
  },
  async (data) => {
    console.log(`New order: ${data.orderId}, total: ${data.total}`)
  },
)
```

**2\. 开始消费:**

```typescript
import { consumeJetStreams } from '@crossdelta/cloudevents'

consumeJetStreams({
  streams: ['ORDERS'],
  consumer: 'my-service',
  discover: './src/events/**/*.handler.ts',
})
```

**3\. 从另一处出版:**

```typescript
import { publish } from '@crossdelta/cloudevents'

await publish('order.created', { orderId: 'ord_123', total: 99.99 })
// String API auto-pluralizes the domain token: subject → 'orders.created'
// Use createContract({ type, channel, schema }) for explicit subject control (no pluralization)
```

就是这样。 Handlers是自动发现的,与Zod验证的,并且消息在JetStream中持续存在.

> **流创建** 由您的发展环境或基础设施处理,而不是由服务处理:
> 
> -   发展: [内兹克州](https://docs.nats.io/running-a-nats-service/nats_admin/jetstream_admin/streams), (中文(简体) ), [@ crossdelta/ platform-sdk (英语)](https://www.npmjs.com/package/@crossdelta/platform-sdk),或手动设置
> -   生产:见 [串流设置](https://npmcn.com/package/%40crossdelta/cloudevents/#readme-stream-setup) 下图 `ensureJetStreams()` 基础设施代码的使用

* * *

## 为什么用这个?

| 问题 | 解决方案 |
| --- | --- |
| 重新启动时丢失的消息 | JetStream 持续消息 |
| 零散处理器登记 | 通过光彩图案自动发现 |
| 运行时类型出错 | 用 TypeScript 推论验证 Zod |
| 有毒信息崩溃服务 | DLQ 隔离无效消息 |

* * *

## 核心概念

### 事件类型对事件数据

**重要的区别:**

-   **事件类型** (单位:千美元)`order.created`:生活在云端 **信封** (单位:千美元)`ce.type`) (中文(简体) ). 用于路由和处理器匹配.
-   **事件数据** (单位:千美元)`{ orderId, total }`实际有效载荷。 对 **没有** 包括类型.

```typescript
const Schema = z.object({
  orderId: z.string(), 
})

export default handleEvent(
  {
    schema: Schema,
    type: 'order.created',
  },
  async (data) => { ... }
)
```

### 处理器

放下一个 `*.handler.ts` 任何地方的文件——是自动注册的:

```typescript
// src/events/user-signup.handler.ts
import { z } from 'zod'

const UserSignupSchema = z.object({ 
  email: z.email(),
  name: z.string(),
})

// Export type for use in use-cases
export type UserSignupEvent = z.infer<typeof UserSignupSchema>

export default handleEvent(
  {
    schema: UserSignupSchema,
    type: 'user.signup',
  },
  async (data) => {
    await sendWelcomeEmail(data.email)
  },
)
```

### 出版

```typescript
await publish('order.created', orderData)
```

### 串流设置

> **仅供基础设施使用** - 服务不应该叫这个功能.

在基础设施设置中创建流( Pulumi, 部署脚本) :

```typescript
import { ensureJetStreams } from '@crossdelta/cloudevents'

await ensureJetStreams({
  streams: [
    {
      stream: 'ORDERS',
      subjects: ['order.>'],  // singular — matches contract subjects (order.created, order.completed, …)
      config: {
        maxAge: 7 * 24 * 60 * 60 * 1000, // 7 days retention
        replicas: 3,                       // For HA
      },
    },
  ],
})
```

### 消费

```typescript
import { consumeJetStreams } from '@crossdelta/cloudevents'

// JetStream (recommended) — persistent, retries, exactly-once
consumeJetStreams({
  streams: ['ORDERS'],
  consumer: 'billing',
  discover: './src/events/**/*.handler.ts',
})

// Core NATS — fire-and-forget, simpler
await consumeNatsEvents({
  subject: 'notifications.*',
  consumerName: 'my-service',
  discover: './src/events/**/*.handler.ts',
})

// NATS Core Request-Reply — synchronous validation feedback
// Consumer returns result, publisher awaits it
import { request } from '@crossdelta/cloudevents'
const result = await request('call.received', payload)
// result: { accepted: true, data: { ... } } | { accepted: false, error: '...' }
```

* * *

## 配置

### 环境变量

```bash
NATS_URL=nats://localhost:4222
NATS_USER=myuser        # optional
NATS_PASSWORD=mypass    # optional
```

### 可选内容:Google Cloud Pub/Sub

要使用 Google Cloud Pub/ Sub 而不是 NATS, 请安装可选的依赖 :

```bash
bun add @google-cloud/pubsub
# or
npm install @google-cloud/pubsub
```

那就用那个 `publishToPubSub` 函数 :

```typescript
import { publishToPubSub } from '@crossdelta/cloudevents'

await publishToPubSub('order.created', { orderId: 'ord_123', total: 99.99 })
```

**说明:** `@google-cloud/pubsub` 不被捆绑到此库中以保持软件包大小。 只有当您需要 Google Cloud Pub/ Sub 支持时才安装它 .

### 流出选项

```typescript
await ensureJetStreams({
  streams: [
    {
      // Required
      stream: 'ORDERS',
      subjects: ['order.>'],  // singular — matches contract subjects

      // Optional
      config: {
        maxAge: 7 * 24 * 60 * 60 * 1000,  // Message retention (ms)
        maxBytes: 1024 * 1024 * 1024,       // Reserved at creation, not on write
        replicas: 1,                        // Replication factor
        storage: 'file',                    // 'file' or 'memory'
        retention: 'limits',                // 'limits', 'interest', or 'workqueue'
      },
    },
  ],
  // Optional
  servers: 'nats://localhost:4222',
})
```

### 消费者选择

```typescript
consumeJetStreams({
  // Required
  streams: ['ORDERS'],
  consumer: 'my-service',
  discover: './src/events/**/*.handler.ts',

  // Optional
  servers: 'nats://localhost:4222',
  filterSubjects: ['order.created'],  // Filter specific subjects
  maxDeliver: 5,                       // Retry attempts
  ackWait: 30_000,                     // Timeout per attempt (ms)
  quarantineTopic: 'dlq',              // For poison messages
})
```

* * *

## 高级特性

**通道元数据( 基础设施集成)**  

合同可包括: **频道元数据** 来定义它们属于哪个NATS JetStream流. 这将使基础设施全码工作流程能够实现合同中的流程.

### 与频道的基本合同

```typescript
import { createContract } from '@crossdelta/cloudevents'
import { z } from 'zod'

export const OrdersCreatedContract = createContract({
  type: 'order.created',
  channel: {
    stream: 'ORDERS',
    // subject defaults to 'order.created' if omitted
  },
  schema: z.object({
    orderId: z.string(),
    total: z.number(),
  }),
})
```

### 频道配置

```typescript
interface ChannelConfig {
  stream: string    // JetStream stream name (e.g., 'ORDERS')
  subject?: string  // NATS subject (optional override)
}
```

**主题路线 :**

-   **无 `subject`**: 默认为 `type` —— 说 `order.created` 留下来 `order.created`
-   **与 `subject`**: 使用精确的字符串

> **说明:** `publish('order.x', data)` (字符串 API) 自动拓扑 → `orders.x`合同没有. 配置 JetStream 有单个主题的流流( )`order.>`与合同活动相匹配.

```typescript
// subject = 'order.created' (from type)
createContract({ type: 'order.created', channel: { stream: 'ORDERS' } })

// subject = 'orders.v1.created' (explicit override)
createContract({ type: 'order.created', channel: { stream: 'ORDERS', subject: 'orders.v1.created' } })
```

### 多个事件, 相同流

```typescript
// order.created → ORDERS stream
export const OrdersCreatedContract = createContract({
  type: 'order.created',
  channel: { stream: 'ORDERS' },
  schema: OrderCreatedSchema,
})

// order.updated → ORDERS stream (same stream!)
export const OrdersUpdatedContract = createContract({
  type: 'order.updated',
  channel: { stream: 'ORDERS' },
  schema: OrderUpdatedSchema,
})
```

**结果 :** 两个活动都分享 `ORDERS` 包含主题的流 `order.created` 财务报告和审定财务报表 `order.updated`。 。 。 .

### 基础设施

频道元数据是 **概念** - 它定义了路线,而不是基础设施政策.

**合同定义:**

-   事件类型( E)`order.created`(中文(简体) )
-   流出路由 (`ORDERS`(中文(简体) )
-   主题映射( O)`order.created`(中文(简体) )

**基础设施定义:**

-   留用(7天,14天等)
-   存储( 磁盘、 内存)
-   复制品(1,3等)
-   限制( 最大消息大小等)

**实例基础设施( Pulumi) :**

```typescript
import { collectStreamDefinitions } from './helpers/streams'
import { ensureJetStreams } from '@crossdelta/cloudevents'

// Collect from contracts
const streams = collectStreamDefinitions()
// [{ stream: 'ORDERS', subjects: ['order.created', 'order.updated'] }]

// Materialize with policies
await ensureJetStreams({
  streams: streams.map(({ stream, subjects }) => ({
    stream,
    subjects,
    config: {
      maxAge: 14 * 24 * 60 * 60 * 1000, // 14 days
      replicas: 3,                       // High availability
      storage: 'file',                   // Persistent
    },
  })),
})
```

### 发展与生产

| 环境 | 流线 | 怎么样 | 保留 |
| --- | --- | --- | --- |
| **发展** | 以弗时 | 由服务自动创建 | 无(纪念) |
| **生产** | 持久性 | 通过基础设施实现 | 明确(7d,14d等) |

**关键原则:** 合同定义 **什麽** 财务报告和审定财务报表 **地点**,基础设施定义 **怎么样** 财务报告和审定财务报表 **多长时间**。 。 。 .

### 后向兼容性

频道元数据是 **选项** - 无合同 `channel` 工作如前:

```typescript
// Legacy contract (still works)
export const LegacyContract = createContract({
  type: 'legacy.event',
  schema: LegacySchema,
  // No channel - handler still works
})
```
**共享合同**

当多个服务消耗同一事件时, 使用 `createContract` 创建共享活动合同:

```typescript
// packages/contracts/src/events/orders/created.ts
import { createContract } from '@crossdelta/cloudevents'
import { z } from 'zod'

export const OrderCreatedSchema = z.object({
  orderId: z.string(),
  total: z.number(),
})

export type OrderCreatedData = z.infer<typeof OrderCreatedSchema>

export const OrdersCreatedContract = createContract({
  type: 'order.created',
  schema: OrderCreatedSchema,
})
```

然后导入和用于处理器 :

```typescript
import { handleEvent } from '@crossdelta/cloudevents'
import { OrdersCreatedContract } from '@my-org/contracts'

export default handleEvent(
  OrdersCreatedContract,
  async (data) => {
    // data is typed as OrderCreatedData
    console.log(data.orderId)
  },
)
```

福利:

-   事件计划的唯一真相来源
-   没有重复计划
-   跨服务类别安全

**多用途**

按租户过滤事件 :

```typescript
export default handleEvent({
  type: 'order.created',
  schema: OrderSchema,
  tenantId: 'tenant-a',  // Only process tenant-a events
}, async (data) => { ... })
```
**通关卡处理器( Catch- all)**

使用 `type: '*'` 创建处理器 **全部** 事件,无论类型。 可用于审计记录、衡量标准收集或事件重放.

```typescript
import { handleEvent } from '@crossdelta/cloudevents'
import { z } from 'zod'

export default handleEvent(
  {
    type: '*',
    schema: z.record(z.string(), z.unknown()),
  },
  async (data, context) => {
    await persistAuditEvent({
      eventType: context.eventType,
      source: context.source,
      data,
    })
  },
)
```

**运行行为 :**

-   具体操作人员(`type: 'order.created'`总是优先
-   通配符处理器仅在没有特定处理器匹配时运行
-   每个消费者只推荐一位通配卡管理员
-   与JetStream公司和NATS公司合作

**自定义匹配**

添加自定义过滤逻辑 :

```typescript
export default handleEvent({
  type: 'order.created',
  schema: OrderSchema,
  match: (event) => event.data.total > 100,  // Only high-value orders
}, async (data) => { ... })
```
**目 录**

分相是内相. 对于分布式系统,提供Redis商店:

```typescript
const redisStore = {
  has: (id) => redis.exists(`idem:${id}`),
  add: (id, ttl) => redis.set(`idem:${id}`, '1', 'PX', ttl),
}

await consumeJetStreams({
  streams: ['ORDERS'],
  consumer: 'my-service',
  discover: './src/events/**/*.handler.ts',
  idempotencyStore: redisStore,
})
```
**死信队列**

无效消息被隔离, 不丢失 :

```typescript
await consumeJetStreams({
  streams: ['ORDERS'],
  consumer: 'my-service',
  discover: './src/events/**/*.handler.ts',
  quarantineTopic: 'events.quarantine',  // For malformed messages
  errorTopic: 'events.errors',            // For handler errors
})
```

无法验证的信件到 `quarantineTopic`, 崩溃处理器前往的信息 `errorTopic`。 。 。 .

**HTTP 摄入量( Hono)**

```typescript
import { Hono } from 'hono'
import { cloudEvents } from '@crossdelta/cloudevents'

const app = new Hono()
app.use('/events', cloudEvents({ discover: 'src/events/**/*.handler.ts' }))
```

* * *

## API 密码

| 函数 | 目的 |
| --- | --- |
| `handleEvent(options, handler)` | 创建处理器( 支持) `type: '*'` 通配符) |
| `createContract(options)` | 创建共享事件合同 |
| `ensureJetStreams(options)` | 创建/更新 JetStream 流 |
| `consumeJetStreams(options)` | 从多条溪流相接 |
| `consumeNatsEvents(options)` | 装火和忘记 |
| `publish(type, data)` | 向 JetStream 发布事件( 火灾和遗忘) |
| `request(type, data, opts?)` | 通过 NATS Core 发送命令并等待同步回复 |
| `isNatsConnected()` | 检查NATS 发布器连接是否为实时(用于健康检查) |
| `isConsumerConnected()` | 检查是否有任何NATS的消费者连接是实时的(准备状态探测器) |
| `configureNatsPublisher(config)` | 为出版商单通设置连接选项 |
| `closeConnection()` | 排水并关闭NATS连接(用于CLI工具) |

* * *

## 分布式追踪

何时 `@crossdelta/telemetry` 已安装(可选的对等依赖),微量上下文自动在 NATS 中传播:

-   **发布**: 当前 W3C `traceparent` 作为 CloudEvent 扩展字段注入.
-   **旋转**编号: `traceparent` 在处理器运行之前,在 Async LocalStorage 中提取并恢复扩展.

这使得Grafana的端到端分布痕迹得以实现:

```
Storefront → API Gateway → NATS → Orders Service → NATS → Audit Service
                └── all spans share the same traceId ──┘
```

不需要配置 。 若为 `@crossdelta/telemetry` 没有安装, CloudEvents 正常工作 没有痕量传播.

## 许可证

麻省理工学院 [横跨德尔塔](https://crossdelta.de/)

## 常见问题

### @crossdelta/cloudevents 是什么？

用于类型Script的 CloudEvents 工具包 - Zod验证,处理器发现,NATS JetStream & Core

### 如何安装 @crossdelta/cloudevents？

在项目目录运行 npm install @crossdelta/cloudevents，或使用 pnpm add / yarn add / bun add。它还需要宿主项目提供 peer 依赖：zod（^4\.0\.0）。npm 7 及以上与 pnpm 8 及以上默认会自动安装必需的 peer 依赖，Yarn 需要手动安装。它还提供命令行工具 cloudevents，可以用 npx cloudevents 直接运行。

### @crossdelta/cloudevents 支持 TypeScript 吗？

支持。包内自带 TypeScript 类型声明（package\.json 声明了 types 入口），安装后即可获得类型提示，无需再安装 @types 包。

### @crossdelta/cloudevents 支持 ESM 还是 CommonJS？

同时提供 ES Module 与 CommonJS 入口，import 和 require\(\) 都可以使用，打包工具会自动选择合适的版本。

### @crossdelta/cloudevents 可以商用吗？

@crossdelta/cloudevents 使用 MIT 许可。宽松许可：允许商业使用、修改和再分发，通常只需在分发时保留原版权与许可声明。以上为概括，不构成法律意见，请以许可原文为准。

### @crossdelta/cloudevents 的下载量有多少？

根据 npm 官方下载统计，@crossdelta/cloudevents 在 2026/10/11 – 2026/11/09 这 30 天内被下载 4,041 次，日均约 135 次。下载量会把 CI 构建、镜像同步和作为其他包依赖的安装一并计入，适合横向比较热度，不能直接等同于使用人数。

### @crossdelta/cloudevents 还在维护吗？

最新版本 0\.10\.1 发布于 2026/10/25（16 天前）；至今共发布 71 个版本，其中过去 12 个月发布了 71 个；首次发布于 2025/12/03。发布频率只是参考信号：功能稳定的工具可能很少需要更新，建议结合源码仓库的 issue 和提交记录判断。