# OIP-Message

**Repository Path**: baikuaifw/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**: 4
- **Created**: 2026-10-28
- **Last Updated**: 2026-10-28

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# OIP-Message · Message Distribution Center

> **Document version**: v1.1 | **Updated**: 2026-09-26 | **Status**: Released
> **Related**: [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` is the **L4 message distribution center** of the Open Interconnect Protocol (OIP), registered on `OIP-GoInOne`. `OIP-Message` is both an `OIP-GoInOne` client (connected and managed) and a service provider registered on `OIP-GoInOne` (receiving message delivery requests and providing event subscription).

> Rust implementation (edition 2024; axum + tokio + mongodb + bson + redis + async-nats + reqwest + futures-util), aligned with [OIP-Message-Design.md](../OIP/docs/OIP-Message-Design.md) in the OIP specification repository.

## Core Capabilities

| Capability | Description |
| --- | --- |
| Message caching / consumption | Lease queue (pending / leased / acked / dead) state machine |
| Pull & ACK | `pull` batch leasing + `ack` confirmation; expired leases auto-redelivered |
| Broadcast | Group/organization broadcast, one independent queue copy per member |
| Session history | Paginated per-session query (reverse chronological) |
| Topic subscription | Unicast topic subscribe/unsubscribe, idempotent |
| Event stream | Topic-pattern subscription (limited to a **registry topic** or a **registry domain prefix + wildcard segment**; a bare `oip.events.>` full subscription is **rejected**) + SSE push |
| Offline messages | Cached for offline users, pulled on reconnect |
| TTL archiving | Normal messages archived after 30 days; system messages (notifications) kept forever |
| Storage abstraction | **Local file JSON snapshot (`store.mode=file`)** / MongoDB (`store.mode=mongo`) / in-memory (`store.mode=memory`) |
| Event bus | In-memory / NATS JetStream dual backends (`events.mode`) |
| **Request log ingestion** | Full request logs reported asynchronously by `OIP-GoInOne`, append-only, idempotent by trace_id (M8); persisted in `file` mode |
| **Subject mailbox** | Per-DID aggregation of offline / pending / backlog delivery status (M1); persisted in `file` mode |
| **Reliable write-fanout** | `fanout` (M2): deterministic copy ids `(copy_id, member_did)`, idempotent dedup, no duplicate delivery |
| **Node-to-node event federation** | Federation (M7): cross-cluster event bridging + inbound dedup (approx. Exactly-Once) |
| Rate limiting | Token bucket (`src/ratelimit.rs`: send 100/s, pull 200/s, ack 200/s, history 100/s, broadcast 10/s, events 10/s, plus a shared global 1000/s bucket), applied to send / pull / ack / history / broadcast / subscribe / fanout / event subscribe and stream / request-log ingestion / federation ingest; **keys are built from the authenticated subject (the front door–adjudicated `x-oip-caller-did`)**, excluding client-controlled fields |

## Quick Start

```bash
# 1. Prepare config (see config.example.yaml)
cp config.example.yaml config.yaml
# Edit config.yaml: fill in store / events / goinone.base_url, etc.

# 2. Build / Test / Lint
cargo build --release --locked
cargo fmt --all --check
cargo clippy --all-targets -- -D warnings
cargo test --all-targets   # unit tests + tests/goinone_contract.rs contract integration tests

# 3. Run (override the config path with OIP_MESSAGE_CONFIG)
OIP_MESSAGE_CONFIG=config.yaml cargo run
```

## Continuous Integration

`.github/workflows/ci.yml` runs on push (`master`/`main`) and `pull_request` on `ubuntu-latest` with Rust `1.98.1` (including `rustfmt` and `clippy`), backed by service containers `nats:2.10` (JetStream enabled, `-js`), `redis:7` and `mongo:7`:

```bash
cargo fmt --all --check
cargo clippy --all-targets -- -D warnings
# with OIP_NATS_URI / OIP_REDIS_URI / OIP_MONGO_URI injected, #[ignore] cases run as well
cargo test --all-targets -- --include-ignored
cargo build --release --locked
```

`.github/workflows/contract-check.yml` performs protocol-surface drift verification and is **fail-closed**: if `../OIP/tools/gen_protocol.py` is missing it first tries `git clone --depth 1 https://gitee.com/orginone/OIP.git ../OIP`; if the script is still unavailable after cloning, the job exits with an error (it does not skip). When available it runs `python3 ../OIP/tools/gen_protocol.py --check`, and any drift fails the job. `src/protocol_gen.rs` and `src/events/topics_gen.rs` are generated by that script.

## Configuration

`config.example.yaml` is the configuration template (containing no sensitive values); copy it to a local `config.yaml` and fill in values as needed. `config.yaml` is excluded by `.gitignore` and is not committed. `src/config.rs` parses strictly with `#[serde(deny_unknown_fields)]`; unknown keys are rejected.

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

| Key | Default | Description |
| --- | --- | --- |
| `listen` | (no default, required) | HTTP listen address; the template uses `0.0.0.0:8091` |
| `log_level` | `info` | Log level: `trace` / `debug` / `info` / `warn` / `error` |
| `store.mode` | `memory` | Message storage: `memory` / `file` / `mongo` |
| `store.mongo_uri` | `""` | MongoDB connection string used when `store.mode=mongo` |
| `store.file_path` | `""` | Message snapshot file path used when `store.mode=file` (parent directory created automatically; empty refuses startup) |
| `events.mode` | `memory` | Event bus: `memory` / `nats` |
| `events.nats_uri` | `""` | NATS address used when `events.mode=nats` |
| `events.cluster_id` | `cluster-local` | Id of this cluster, used to deduplicate remote registration in federation |
| `events.replay` | `memory` | SSE resume replay storage: `memory` / `redis` (not listed in the template) |
| `events.replay_uri` | `""` | Redis URI used when `events.replay=redis` (not listed in the template) |
| `queue.lease_ttl_secs` | `60` | Consumption lease duration (seconds) |
| `queue.ttl_days` | `30` | Message TTL (days); system messages (notifications) are kept forever |
| `queue.max_attempts` | `3` | Maximum delivery attempts before entering Dead |
| `queue.pull_limit` | `20` | Batch pull limit |
| `goinone.base_url` | `""` | `OIP-GoInOne` node address (service registration and push notification; left empty it is skipped) |
| `goinone.auth_token` | `""` | Internal JWT issued by `OIP-GoInOne` (sensitive) |
| `env` | `development` | Runtime env: `production` / `development`; with `production` and no `internal_token`, startup is **refused** (fail-closed) |
| `internal_token` | `""` | Inbound internal token (service-to-service gate, sensitive): non-empty enables it, then all endpoints except the health probes MUST carry `x-oip-internal-token` or `Authorization: Bearer <token>`; MUST equal `OIP-GoInOne`'s `message.auth_token` (also injectable via `OIP_MESSAGE_INTERNAL_TOKEN`) |

The configuration file path can be overridden with `OIP_MESSAGE_CONFIG`; the default is `config.yaml`.

## Persistence Modes

`store.mode` decides how **stateful** data (messages, request logs, mailbox) is persisted:

| Mode | Storage | Use case | Notes |
| --- | --- | --- | --- |
| `memory` | In-process | Unit tests / local demo | Lost on restart |
| `file` | Local filesystem (JSON snapshot) | Production for a single cluster with no external dependencies | Messages → `store.file_path`; request log/mailbox → `request-log.json` / `mailbox.json` in the same directory (atomic write via temp file + rename, crash-safe, recovers across restarts) |
| `mongo` | MongoDB | Multi-instance / large-scale horizontal scaling | Messages stored in the `messages` / `conversations` / `subscriptions` collections |

`events.mode` controls the **event bus** independently: `memory` (in-process) / `nats` (NATS JetStream). SSE resume replay is decided by `events.replay`: `memory` or `redis`.

The production combination is `store.mode=file` (local-file persistence, single cluster / single instance, crash-safe atomic writes) + `events.mode=nats`; switch to `store.mode=mongo` when multi-instance horizontal scaling is required; a `memory` combination loses data on restart. With `store.mode=file` and an empty `store.file_path`, state construction fails explicitly and refuses startup.

## API Endpoints

`build()` in `src/router.rs` is the single assembly point for all endpoints.

**Subject attribution (security invariant)**: for identity-semantic endpoints (`send` / `pull` / `ack` / `history` / `broadcast` / `fanout` / `subscribe` / `unsubscribe` / `events/subscribe` / `mailbox/{did}`), the calling subject is taken **from the authenticated state** — the front door `OIP-GoInOne` injects the `x-oip-caller-did` header it adjudicated after signature verification, **overriding** `from_id` / `recipient_did` / `subscriber_did` / `did` in the body / query / path (those fields are **targets**, not identities). When the header is absent (direct calls, subject not adjudicated) these endpoints return `OIP_4011`. `federation/ingest`, `request-log` (ingest / query) and `federation/peers` have no subject semantics and are served in a "service-to-service direct" form that does not require the header.

| Method | Path | Description |
| --- | --- | --- |
| POST | `/oip/v1/message/send` | Send a peer-to-peer message |
| GET | `/oip/v1/message/pull` | Pull leased messages |
| POST | `/oip/v1/message/ack` | Acknowledge consumption |
| GET | `/oip/v1/message/history` | Paginated session history |
| POST | `/oip/v1/message/broadcast` | Group broadcast |
| POST | `/oip/v1/message/fanout` | Reliable write-fanout delivery (M2, idempotent) |
| POST | `/oip/v1/message/subscribe` | Topic subscribe |
| POST | `/oip/v1/message/unsubscribe` | Topic unsubscribe |
| POST | `/oip/v1/message/events/subscribe` | Event subscribe (topic pattern) |
| POST | `/oip/v1/message/events/unsubscribe` | Event unsubscribe |
| GET | `/oip/v1/message/events/stream/:sub_id` | SSE event stream |
| POST | `/oip/v1/message/request-log` | Request log ingestion (single/batch) |
| GET | `/oip/v1/message/request-log/query` | Request log query |
| GET | `/oip/v1/message/mailbox/:did` | Subject mailbox delivery status |
| POST | `/oip/v1/message/federation/ingest` | Federation inbound (M7, dedup by event id); inbound `id` is validated: **non-empty, ≤128 chars, no control characters**, otherwise `4001` (the `id` doubles as dedup key / SSE frame field / replay cursor) |
| GET | `/oip/v1/message/federation/peers` | Registered remote clusters (observability) |
| GET | `/healthz` | Liveness probe |
| GET | `/readyz` | Readiness probe |

## Error Codes

`OipErrorCode::as_str()` in `src/error.rs` defines the external error codes and `http_status()` defines the HTTP mapping.

| Code | HTTP | Meaning |
| --- | --- | --- |
| `OIP_4001` | 400 | Invalid parameter (including an illegal `type_name`) |
| `OIP_4011` | 401 | Proof / token validation failed |
| `OIP_4031` | 403 | Rejection (boolean blindness; no reason exposed) |
| `OIP_4041` | 404 | Not found |
| `OIP_4094` | 409 | State conflict (including duplicate ACK) |
| `OIP_4291` | 429 | Rate limited |
| `OIP_5001` | 500 | Internal error (generic message) |
| `OIP_5031` | 503 | Service unavailable |

## Directory Structure

```text
OIP-Message/
├── Cargo.toml / Cargo.lock   # manifest and locked dependencies (builds use --locked)
├── config.example.yaml       # config template (config.yaml is never committed)
├── Dockerfile                # production image (multi-stage, rust:1.98-bookworm → debian:bookworm-slim)
├── docker-compose.yml        # local orchestration
├── deny.toml                 # cargo-deny dependency policy
├── README.md / README.en.md / AGENTS.md
├── .github/workflows/        # ci.yml (build/test/service containers), contract-check.yml (protocol-surface drift gate)
├── tests/
│   └── goinone_contract.rs   # OIP-GoInOne delivery contract integration tests
└── src/
    ├── main.rs          # binary entry (thin shell)
    ├── lib.rs           # library entry (module declarations; exports Config / AppState)
    ├── client/          # OIP-GoInOne client (service registration)
    ├── clock.rs         # time abstraction (SystemClock / ManualClock)
    ├── config.rs        # config loading & validation
    ├── dispatch.rs      # dispatching (peer/broadcast) & topic registry
    ├── error.rs         # OIP error codes & unified error
    ├── events/          # CloudEvent / subscription registry / event consumer / NATS bus / resume replay / topic registry
    ├── federation.rs    # node-to-node event federation (M7, cross-cluster bridging + inbound dedup)
    ├── handlers/        # HTTP handlers (events / health / message / system)
    ├── ids.rs           # message / subscription id generation
    ├── mailbox.rs       # request log ingestion (M8) + subject mailbox (M1)
    ├── protocol_gen.rs  # protocol-surface registry (generated by gen_protocol.py)
    ├── queue.rs         # lease queue logic
    ├── ratelimit.rs     # token bucket rate limiting
    ├── response.rs      # unified OperateResult response
    ├── router.rs        # route assembly + router integration tests
    ├── runtime.rs       # runtime assembly (background maintenance tasks)
    ├── sse.rs           # SSE stream wrapper
    ├── state.rs         # global state and storage/event backend selection
    ├── store/           # message storage abstraction (file / mongo / memory)
    └── types/           # message types (23-category Chinese registry) / queue models
```

## Message Types

Message `type_name` uses the complete 23-category Chinese registry of OIP §14.3 (`ChatMessageType::ALL`: text/web page/image/video/voice/file/recall/notification/forward/name card/location/task/group activity/phrase/Markdown/announcement/entity/agent message/voice call/video call/solitaire/card/red packet), aligned with the four frontend projects (`OIP-Desktop` / `OIP-Web` / `OIP-Mobile` / `OIP-Lite`); sending an illegal type returns `OIP_4001`.

## Testing

| Layer | Location | Description |
| --- | --- | --- |
| Unit tests | `#[cfg(test)]` in each module | lease state machine, ACK idempotency, broadcast copies, topic-subscription idempotency, TTL archiving, request log and mailbox |
| Router integration tests | `#[cfg(test)]` in `src/router.rs` | real HTTP requests via `tower::ServiceExt::oneshot` (including a file-mode restart-recovery case) |
| Contract integration tests | `tests/goinone_contract.rs` | `OIP-GoInOne` delivery contract (11 cases) |
| External-dependency cases | `#[ignore]` + `OIP_NATS_URI` / `OIP_REDIS_URI` / `OIP_MONGO_URI` | require a real NATS JetStream / Redis / MongoDB; print a skip notice and return when unset. CI injects all three variables and runs with `--include-ignored` |

## Security & Ops

- `goinone.auth_token` is injected via `config.yaml` (gitignored, never committed) and never hardcoded
- Inbound internal token gate (defense in depth): once `internal_token` is configured, all endpoints except `/healthz` and `/readyz` MUST carry `x-oip-internal-token` or `Authorization: Bearer <token>`, otherwise `OIP_4011` (no distinction between "missing" and "wrong token"); with `env=production` and no token, startup is **refused** (fail-closed)
- Subject attribution: identity endpoints take the calling subject from the front door–adjudicated `x-oip-caller-did`, **never** trusting self-reported identity in request parameters; absent header → `OIP_4011`
- Event subscription wildcard convergence: only a registry topic or a domain prefix + wildcard segment is allowed; a bare `oip.events.>` full subscription is rejected (`OIP_4001`)
- Background maintenance reclaims expired leases and archives TTL messages every minute with batch caps
- send / pull / ack / history / broadcast / subscribe / fanout / event subscribe and stream / request-log ingestion / federation ingest are rate-limited by token bucket and share the global 1000/s bucket (`src/ratelimit.rs`), with keys built from the authenticated subject; `unsubscribe`, `request-log/query`, `mailbox`, `federation/peers` and the health probes carry no per-endpoint quota
- 500-class errors return generic messages; full exceptions go only to internal logs

## Docs

- Design (OIP specification repository): [OIP-Message-Design.md](../OIP/docs/OIP-Message-Design.md)
- Protocol specification (OIP specification repository): [OIP-Protocol.md](../OIP/docs/OIP-Protocol.md)
- Testing (OIP specification repository): [OIP-Test-Strategy.md](../OIP/docs/OIP-Test-Strategy.md)
- Protocol-surface generator (OIP specification repository): [gen_protocol.py](../OIP/tools/gen_protocol.py)
- Conventions: [AGENTS.md](AGENTS.md)
- License: MIT