# NATS 事件

NATS JetStream 事件主题、负载格式和集成模式的完整参考。

## 前提条件

- 您所用语言的 NATS 客户端库
- NATS Server 2.10+ 且已启用 JetStream
- 到 NATS 集群的网络访问

## 事件架构

```mermaid
graph TB
    subgraph 发布者
        SPI[SPI 服务]
        DICT[Dict 服务]
        SETTLE[Settlement 服务]
    end

    subgraph NATS JetStream
        S1[MONETARIE_SPI<br/>monetarie.spi.>]
        S2[MONETARIE_DICT<br/>monetarie.dict.>]
        S3[MONETARIE_SETTLEMENT<br/>monetarie.settlement.>]
        S4[MONETARIE_CORE<br/>monetarie.core.>]
        S5[MONETARIE_AUDIT<br/>monetarie.audit.>]
        S6[MONETARIE_DLQ<br/>monetarie.dlq.>]
    end

    subgraph 消费者
        WORKERS[PIX Workers]
        CORE[Core Banking]
        CCS[CCS 服务]
    end

    SPI --> S1
    DICT --> S2
    SETTLE --> S3
    S1 --> WORKERS
    S1 --> CORE
    S4 --> WORKERS
    S1 --> CCS
```

## SPI 事件

| 主题 | 描述 |
|------|------|
| `monetarie.spi.transaction.created` | 新交易已创建 |
| `monetarie.spi.transaction.processing` | 交易处理中（ACSP） |
| `monetarie.spi.transaction.settled` | 交易已清算（STLD） |
| `monetarie.spi.transaction.rejected` | 交易被拒绝（RJCT） |
| `monetarie.spi.transaction.returned` | 交易已退回（RTRN） |
| `monetarie.spi.transaction.cancelled` | 交易已取消（CANC） |
| `monetarie.spi.inbound.received` | 从 BACEN 接收到入站支付 |
| `monetarie.spi.outbound.sent` | 出站支付已发送到 BACEN |
| `monetarie.spi.balance.updated` | 余额已变更 |

### SPI 事件负载

```json
{
  "type": "transaction.settled",
  "source": "pix",
  "published_at": "2026-02-09T14:30:00Z",
  "data": {
    "transaction_id": "uuid",
    "end_to_end_id": "E12345678202602091430abcdef123456",
    "message_type": "pacs.008",
    "amount": 10000,
    "status": "STLD",
    "direction": "OUTBOUND",
    "debtor_ispb": "12345678",
    "creditor_ispb": "87654321",
    "operation_time": "2026-02-09T14:30:00.000Z",
    "settlement_time": "2026-02-09T14:30:01.200Z"
  }
}
```

## DICT 事件

| 主题 | 描述 |
|------|------|
| `monetarie.dict.keys.created` | PIX 密钥已注册 |
| `monetarie.dict.keys.deleted` | PIX 密钥已删除 |
| `monetarie.dict.keys.claimed` | 所有权认领已发起 |
| `monetarie.dict.keys.ported` | 密钥可携性已完成 |
| `monetarie.dict.cid.synced` | CID 同步事件 |

## Core Banking 事件

### Core -> PIX

| 主题 | 描述 |
|------|------|
| `monetarie.core.pix.payment_request` | 创建出站支付 |
| `monetarie.core.pix.return_request` | 创建退回（pacs.004） |
| `monetarie.core.pix.key_create` | 注册 PIX 密钥 |
| `monetarie.core.pix.key_delete` | 删除 PIX 密钥 |
| `monetarie.core.pix.dict_lookup` | DICT 密钥查询 |
| `monetarie.core.pix.balance_inquiry` | 余额查询 |

### Core -> PIX 负载示例

```json
{
  "event": "payment_request",
  "correlation_id": "uuid",
  "data": {
    "amount": 10000,
    "creditor_key": "email@example.com",
    "debtor_account": "12345",
    "debtor_ispb": "12345678",
    "description": "Payment for services"
  }
}
```

::: warning 路由键
`"event"` 字段是 CoreEventProcessor 的路由键。如果缺失或拼写错误，消息会被静默确认而不处理。
:::

## 请求/回复（非 JetStream）

| 主题 | 方向 | 用途 |
|------|------|------|
| `dict.lookup.request` | Core <-> PIX | DICT 密钥查询 |
| `dict.api.request` | Core <-> PIX | 认领/MED/违规 |

这些主题在 JetStream 范围之外，以避免流拦截。

## 消费事件

### NATS 客户端示例（Go）

```go
js, _ := nc.JetStream()

sub, _ := js.Subscribe("monetarie.spi.transaction.>", func(msg *nats.Msg) {
    var event map[string]interface{}
    json.Unmarshal(msg.Data, &event)

    // 处理事件
    fmt.Printf("Event: %s\n", event["type"])

    // 确认
    msg.Ack()
}, nats.Durable("my-consumer"))
```

### NATS 客户端示例（Node.js）

```javascript
const { connect, JSONCodec } = require("nats")

const nc = await connect({ servers: "10.10.40.5:4222" })
const js = nc.jetstream()
const jc = JSONCodec()

const sub = await js.subscribe("monetarie.spi.transaction.>", {
  durable: "my-consumer"
})

for await (const msg of sub) {
  const event = jc.decode(msg.data)
  console.log(`Event: ${event.type}`)
  msg.ack()
}
```

## 消息头

所有 NATS 消息包含：

| 头 | 描述 |
|----|------|
| `trace_id` | 分布式追踪 ID 用于日志关联 |
| `Nats-Msg-Id` | 去重 ID |
| `published_at` | ISO 8601 时间戳 |

## 预期结果

集成 NATS 事件后：

- 从 PIX 流实时消费事件
- Core Banking 事件双向流通
- 通过 `trace_id` 头实现分布式追踪
- 持久消费者确保至少一次投递