# OIP-Message

**Repository Path**: orginone/OIP-Message

## Basic Information

- **Project Name**: OIP-Message
- **Description**: OIP- Message是奥集能开放架构的 消息分发中心，注册在 GoInOne上。OIP- Message既是 GoInOne的客户端（连上GoInOne、接受管理），又向 GoInOne注册为服务提供方（接收消息投递请求、提供事件订阅能力）。
- **Primary Language**: Rust
- **License**: MIT
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 2
- **Created**: 2026-09-05
- **Last Updated**: 2026-10-18

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# OIP-Message · 消息分发中心

> **文档版本**：v1.1 | **更新**：2026-09-26 | **状态**：正式发布
> **关联**：[OIP-Message-Design.md](../OIP/docs/OIP-Message-Design.md) · [OIP-Protocol.md](../OIP/docs/OIP-Protocol.md) · [OIP.md](../OIP/docs/OIP.md)

`OIP-Message` 是开放互联协议（Open Interconnect Protocol, OIP）的 **L4 消息分发中心**，注册在 `OIP-GoInOne` 上。`OIP-Message` 既是 `OIP-GoInOne` 的客户端（连上 `OIP-GoInOne`、接受管理），又向 `OIP-GoInOne` 注册为服务提供方（接收消息投递请求、提供事件订阅能力）。

> Rust 实现（edition 2024；axum + tokio + mongodb + bson + redis + async-nats + reqwest + futures-util），对齐 OIP 规范仓库中的设计文档 [OIP-Message-Design.md](../OIP/docs/OIP-Message-Design.md)。

## 核心能力

| 能力 | 说明 |
| --- | --- |
| 消息缓存/消费 | 租约队列（pending / leased / acked / dead）状态机 |
| 拉取与 ACK | `pull` 批量租约 + `ack` 确认，租约过期自动重投 |
| 广播 | 群组/组织广播，每成员独立队列副本 |
| 会话历史 | 按会话分页查询（时间倒序） |
| 主题订阅 | 单播主题订阅/退订，幂等 |
| 事件流 | 主题模式订阅（限**注册表内具体主题**或**域前缀 + 通配段**；**禁止**裸 `oip.events.>` 全量订阅）+ SSE 实时推送 |
| 离线消息 | 离线用户消息缓存，上线后拉取 |
| TTL 归档 | 普通消息 30 天归档；系统消息（通知）永久保存 |
| 存储抽象 | **本地文件 JSON 快照（`store.mode=file`）** / MongoDB（`store.mode=mongo`） / 内存（`store.mode=memory`） |
| 事件总线 | 内存 / NATS JetStream 双实现（`events.mode`） |
| **请求日志归集** | `OIP-GoInOne` 全量请求日志异步上报，只追加、按 trace_id 幂等（M8）；`file` 模式下落盘 |
| **主体信箱** | 按 DID 汇总离线/待投递/积压投递状态（M1）；`file` 模式下落盘 |
| **写扩散可靠投递** | `fanout`（M2）：`(copy_id, member_did)` 确定性副本 id，幂等去重不重复投递 |
| **节点间事件联邦** | Federation（M7）：跨集群事件桥接 + 入站去重（Exactly-Once 近似） |
| 限流 | 令牌桶（`src/ratelimit.rs`：send 100/s、pull 200/s、ack 200/s、history 100/s、broadcast 10/s、events 10/s，并共享全局 1000/s 桶），配额落在 send / pull / ack / history / broadcast / subscribe / fanout / 事件订阅与事件流 / 请求日志上报 / 联邦入站；**限流键按认证主体（前门裁定的 `x-oip-caller-did`）构造**，剔除客户端可控字段 |

## 快速开始

```bash
# 1. 准备配置（模板见 config.example.yaml）
cp config.example.yaml config.yaml
# 编辑 config.yaml：填写 store / events / goinone.base_url 等

# 2. 构建 / 测试 / Lint
cargo build --release --locked
cargo fmt --all --check
cargo clippy --all-targets -- -D warnings
cargo test --all-targets   # 单元测试 + tests/goinone_contract.rs 契约集成测试

# 3. 运行（可用 OIP_MESSAGE_CONFIG 覆盖配置路径）
OIP_MESSAGE_CONFIG=config.yaml cargo run
```

## 持续集成

`.github/workflows/ci.yml` 在 push（`master`/`main`）与 `pull_request` 时执行，运行于 `ubuntu-latest`，工具链 Rust `1.98.1`（含 `rustfmt`、`clippy`），并以服务容器提供 `nats:2.10`（启用 JetStream，`-js`）、`redis:7`、`mongo:7`：

```bash
cargo fmt --all --check
cargo clippy --all-targets -- -D warnings
# 注入 OIP_NATS_URI / OIP_REDIS_URI / OIP_MONGO_URI 后连同 #[ignore] 用例一并执行
cargo test --all-targets -- --include-ignored
cargo build --release --locked
```

`.github/workflows/contract-check.yml` 执行契约面漂移校验，且为 **fail-closed**：若同级目录不存在 `../OIP/tools/gen_protocol.py`，先尝试 `git clone --depth 1 https://gitee.com/orginone/OIP.git ../OIP`；克隆后仍不可用则报错退出（非跳过）。可用时执行 `python3 ../OIP/tools/gen_protocol.py --check`，任何漂移即失败。`src/protocol_gen.rs`、`src/events/topics_gen.rs` 由该脚本生成。

## 配置

`config.example.yaml` 为配置模板（不含敏感值），复制为本地 `config.yaml` 后按需填写；`config.yaml` 由 `.gitignore` 排除，不入库。`src/config.rs` 以 `#[serde(deny_unknown_fields)]` 严格解析，未列出的键会被拒绝。

```bash
cp config.example.yaml config.yaml
```

| 键 | 默认值 | 说明 |
| --- | --- | --- |
| `listen` | （无默认，必填） | HTTP 监听地址；模板为 `0.0.0.0:8091` |
| `log_level` | `info` | 日志级别：`trace` / `debug` / `info` / `warn` / `error` |
| `store.mode` | `memory` | 消息存储：`memory` / `file` / `mongo` |
| `store.mongo_uri` | `""` | `store.mode=mongo` 时的 MongoDB 连接串 |
| `store.file_path` | `""` | `store.mode=file` 时的消息快照文件路径（父目录自动创建；留空则拒绝启动） |
| `events.mode` | `memory` | 事件总线：`memory` / `nats` |
| `events.nats_uri` | `""` | `events.mode=nats` 时的 NATS 地址 |
| `events.cluster_id` | `cluster-local` | 本集群 id，用于联邦远端登记去重 |
| `events.replay` | `memory` | SSE 断点续传回放存储：`memory` / `redis`（模板未列出） |
| `events.replay_uri` | `""` | `events.replay=redis` 时的 Redis 连接串（模板未列出） |
| `queue.lease_ttl_secs` | `60` | 消费租约时长（秒） |
| `queue.ttl_days` | `30` | 消息 TTL（天）；系统消息（通知）永久保存 |
| `queue.max_attempts` | `3` | 最大投递尝试次数，超过后进入 Dead |
| `queue.pull_limit` | `20` | 批量拉取上限 |
| `goinone.base_url` | `""` | `OIP-GoInOne` 节点地址（服务注册与推送通知；留空则跳过） |
| `goinone.auth_token` | `""` | `OIP-GoInOne` 签发的内部 JWT（敏感项） |
| `env` | `development` | 运行环境：`production` / `development`；`production` 下未配置 `internal_token` 则**拒绝启动**（fail-closed） |
| `internal_token` | `""` | 入站内部令牌（服务间门禁，敏感项）：非空即启用，除健康探针外全部端点 MUST 携带 `x-oip-internal-token` 或 `Authorization: Bearer <token>`；MUST 与 `OIP-GoInOne` 的 `message.auth_token` 同值（亦可用 `OIP_MESSAGE_INTERNAL_TOKEN` 注入） |

配置文件路径可用 `OIP_MESSAGE_CONFIG` 覆盖，默认 `config.yaml`。

## 持久化模式

`store.mode` 决定消息、请求日志、信箱等**有状态**数据的落盘方式：

| 模式 | 存储位置 | 适用场景 | 说明 |
| --- | --- | --- | --- |
| `memory` | 进程内 | 单元测试 / 本地演示 | 重启即丢失 |
| `file` | 本地文件系统（JSON 快照） | 单集群、无外部依赖的生产部署 | 消息 → `store.file_path`；请求日志/信箱 → 同目录 `request-log.json` / `mailbox.json`（临时文件 + rename 原子写，崩溃安全，可跨重启恢复） |
| `mongo` | MongoDB | 多实例 / 大规模横向扩展 | 消息落 `messages` / `conversations` / `subscriptions` 三集合 |

`events.mode` 单独控制**事件总线**：`memory`（进程内）/ `nats`（NATS JetStream）。SSE 断点续传回放由 `events.replay` 决定：`memory` 或 `redis`。

生产组合为 `store.mode=file`（本地文件持久化，单集群单实例、原子写崩溃安全）+ `events.mode=nats`；需要多实例水平扩展时切换 `store.mode=mongo`；`memory` 组合重启即丢数据。`store.mode=file` 而未配置 `store.file_path` 时，状态构建显式报错并拒绝启动。

## API 端点

`src/router.rs` 的 `build()` 为全部端点的唯一装配点。

**主体归属（安全不变量）**：以身份为语义的端点（`send` / `pull` / `ack` / `history` / `broadcast` / `fanout` / `subscribe` / `unsubscribe` / `events/subscribe` / `mailbox/{did}`）的调用主体**取自认证态**——由前门 `OIP-GoInOne` 验签裁定的 `x-oip-caller-did` 头注入，**覆盖**请求体 / 查询串 / 路径中的 `from_id` / `recipient_did` / `subscriber_did` / `did`（这些字段只作**目标**、不作身份）。该头缺失（直连调用、主体未裁定）时上述端点返回 `OIP_4011`。`federation/ingest`、`request-log`（上报 / 查询）与 `federation/peers` 无主体语义，以「服务间直连形态」受理，不要求该头。

| 方法 | 路径 | 说明 |
| --- | --- | --- |
| POST | `/oip/v1/message/send` | 发送点对点消息 |
| GET | `/oip/v1/message/pull` | 拉取租约消息 |
| POST | `/oip/v1/message/ack` | 确认消费 |
| GET | `/oip/v1/message/history` | 会话历史分页 |
| POST | `/oip/v1/message/broadcast` | 群组广播 |
| POST | `/oip/v1/message/fanout` | 写扩散可靠投递（M2，幂等） |
| POST | `/oip/v1/message/subscribe` | 主题订阅 |
| POST | `/oip/v1/message/unsubscribe` | 主题退订 |
| POST | `/oip/v1/message/events/subscribe` | 事件订阅（主题模式） |
| POST | `/oip/v1/message/events/unsubscribe` | 事件退订 |
| GET | `/oip/v1/message/events/stream/:sub_id` | SSE 事件流 |
| POST | `/oip/v1/message/request-log` | 请求日志上报（单条/批量） |
| GET | `/oip/v1/message/request-log/query` | 请求日志查询 |
| GET | `/oip/v1/message/mailbox/:did` | 主体信箱投递状态 |
| POST | `/oip/v1/message/federation/ingest` | 联邦入站（M7，按 event id 去重）；入站 `id` 校验：**非空、≤128 字符、不含控制字符**，不合规 `4001`（`id` 同时是去重键 / SSE 帧字段 / 回放游标） |
| GET | `/oip/v1/message/federation/peers` | 已登记远端集群（观测） |
| GET | `/healthz` | 存活探针 |
| GET | `/readyz` | 就绪探针 |

## 错误码

`src/error.rs` 的 `OipErrorCode::as_str()` 定义对外错误码，`http_status()` 定义 HTTP 映射。

| 错误码 | HTTP | 含义 |
| --- | --- | --- |
| `OIP_4001` | 400 | 参数非法（含非法 `type_name`） |
| `OIP_4011` | 401 | 举证/令牌校验失败 |
| `OIP_4031` | 403 | 拒绝（布尔盲化，不含原因） |
| `OIP_4041` | 404 | 资源不存在 |
| `OIP_4094` | 409 | 状态冲突（含重复 ACK） |
| `OIP_4291` | 429 | 限流 |
| `OIP_5001` | 500 | 服务内部错误（通用消息） |
| `OIP_5031` | 503 | 服务不可用 |

## 目录结构

```text
OIP-Message/
├── Cargo.toml / Cargo.lock   # 工程清单与锁定依赖（构建走 --locked）
├── config.example.yaml       # 配置模板（config.yaml 不入库）
├── Dockerfile                # 生产镜像（多阶段构建，rust:1.98-bookworm → debian:bookworm-slim）
├── docker-compose.yml        # 本地编排
├── deny.toml                 # cargo-deny 依赖策略
├── README.md / README.en.md / AGENTS.md
├── .github/workflows/        # ci.yml（构建/测试/服务容器）、contract-check.yml（协议面漂移门禁）
├── tests/
│   └── goinone_contract.rs   # OIP-GoInOne 投递契约集成测试
└── src/
    ├── main.rs          # 二进制入口（薄壳）
    ├── lib.rs           # 库入口（模块声明，导出 Config / AppState）
    ├── client/          # OIP-GoInOne 客户端（服务注册）
    ├── clock.rs         # 时间抽象（SystemClock / ManualClock）
    ├── config.rs        # 配置加载与校验
    ├── dispatch.rs      # 分发（点对点/广播）与主题注册表
    ├── error.rs         # OIP 错误码与统一错误
    ├── events/          # CloudEvent / 订阅注册表 / 事件消费者 / NATS 总线 / 断点续传回放 / 主题规范表
    ├── federation.rs    # 节点间事件联邦（M7，跨集群桥接 + 入站去重）
    ├── handlers/        # HTTP 处理器（events / health / message / system）
    ├── ids.rs           # 消息 / 订阅 ID 生成
    ├── mailbox.rs       # 请求日志归集（M8）+ 主体信箱（M1）
    ├── protocol_gen.rs  # 协议面规范表（由 gen_protocol.py 生成）
    ├── queue.rs         # 租约队列逻辑
    ├── ratelimit.rs     # 令牌桶限流
    ├── response.rs      # 统一 OperateResult 响应
    ├── router.rs        # 路由装配 + 路由层集成测试
    ├── runtime.rs       # 服务运行时装配（后台维护任务）
    ├── sse.rs           # SSE 流封装
    ├── state.rs         # 全局状态与存储/事件后端选型
    ├── store/           # 消息存储抽象（file / mongo / memory）
    └── types/           # 消息类型（23 类中文注册表）/ 队列模型
```

## 消息类型

消息 `type_name` 使用 OIP §14.3 的完整 23 类中文注册表（`ChatMessageType::ALL`：文本/网页/图片/视频/语音/文件/撤回/通知/转发/名片/位置/任务/群动态/词组/Markdown/声明/实体/智能体消息/语音通话/视频通话/接龙/卡片/红包），与前端四项目（`OIP-Desktop` / `OIP-Web` / `OIP-Mobile` / `OIP-Lite`）对齐；非法类型发送返回 `OIP_4001`。

## 测试

| 层级 | 位置 | 说明 |
| --- | --- | --- |
| 单元测试 | 各模块 `#[cfg(test)]` | 租约状态机、ACK 幂等、广播副本、主题订阅幂等、TTL 归档、请求日志与信箱 |
| 路由集成测试 | `src/router.rs` 的 `#[cfg(test)]` | 经 `tower::ServiceExt::oneshot` 发起真实 HTTP 请求断言（含 file 模式跨重启恢复用例） |
| 契约集成测试 | `tests/goinone_contract.rs` | `OIP-GoInOne` 投递契约（11 个用例） |
| 外部依赖用例 | `#[ignore]` + `OIP_NATS_URI` / `OIP_REDIS_URI` / `OIP_MONGO_URI` | 需真实 NATS JetStream / Redis / MongoDB；未设置时打印跳过信息并返回。CI 注入三个变量并以 `--include-ignored` 执行 |

## 安全与运维

- `goinone.auth_token` 经 `config.yaml`（`.gitignore` 排除，不入库）注入，不硬编码
- 入站内部令牌门（纵深防御）：配置 `internal_token` 后，除 `/healthz`、`/readyz` 外全部端点 MUST 携带 `x-oip-internal-token` 或 `Authorization: Bearer <token>`，否则 `OIP_4011`（不区分「未携带」与「令牌错误」）；`env=production` 而未配置该令牌时**拒绝启动**（fail-closed）
- 主体归属：身份类端点的调用主体取自前门裁定的 `x-oip-caller-did`，**不采信**请求参数中的自报身份；缺失即 `OIP_4011`
- 事件订阅通配收敛：仅允许注册表内具体主题或域前缀 + 通配段，裸 `oip.events.>` 全量订阅被拒（`OIP_4001`）
- 后台维护任务每分钟执行租约回收与 TTL 归档，单次批量上限防过载
- send / pull / ack / history / broadcast / subscribe / fanout / 事件订阅与事件流 / 请求日志上报 / 联邦入站均受令牌桶限流，并共享全局 1000/s 桶（`src/ratelimit.rs`），限流键按认证主体构造；`unsubscribe`、`request-log/query`、`mailbox`、`federation/peers` 与健康探针不设端点配额
- 500 类错误对外返回通用信息，异常详情只进内部日志

## 文档

- 设计文档（OIP 规范仓库）：[OIP-Message-Design.md](../OIP/docs/OIP-Message-Design.md)
- 协议规范（OIP 规范仓库）：[OIP-Protocol.md](../OIP/docs/OIP-Protocol.md)
- 测试策略（OIP 规范仓库）：[OIP-Test-Strategy.md](../OIP/docs/OIP-Test-Strategy.md)
- 协议面生成脚本（OIP 规范仓库）：[gen_protocol.py](../OIP/tools/gen_protocol.py)
- 开发约定：[AGENTS.md](AGENTS.md)
- License: MIT