站内信（@weifuwujs/messages）

用户通知的最小完整形态：服务端发送（幂等） + REST 收件箱 + WebSocket 实时推送 + 断线 since 补投，memory / sqlite / postgres 三驱动共用同一套契约测试。适合站内信、系统通知、 "任务完成"提醒这类"写入一次、本人可查、要实时"的场景。

不适合的：IM 聊天（双向会话、打字状态、已读回执）、服务间消息总线（用 @weifuwujs/tasks 或 MQ）、给非登录用户群发（消息挂在用户上，先有账号）。

十分钟接入

最快的路径是脚手架 + 包内叠加模板：

npx @weifuwujs/weifuwu init my-app
cd my-app
cp -r "$(npm root)/@weifuwujs/auth/templates/react-router-auth/." .
cp -r "$(npm root)/@weifuwujs/messages/templates/react-router-messages/." .
npx @weifuwujs/auth migrate
npx @weifuwujs/messages migrate   # 欢迎信 + 徽标 + /notifications 页 + WS 实时
npm run dev

手工接入只有四步（选驱动 → createMessages → install(app) → send）：

import { createApp } from '@weifuwujs/weifuwu'
import { createMessages } from '@weifuwujs/messages'
import { sqliteMessageStore } from '@weifuwujs/messages/sqlite'

const messages = createMessages({
  store: sqliteMessageStore('data/messages.db'), // 构造即迁移（幂等）
})

const app = createApp()
messages.install(app) // REST + WebSocket 一条线（都要求登录，未认证 401），并随 app.close() 收连接

await messages.send({
  userId: 'user-1',
  kind: 'welcome',
  title: '欢迎使用',
  body: '从这条消息开始。',
  idempotencyKey: 'welcome:user-1', // 重复发送只入库一次，不覆盖
})

最小可跑示例（零依赖，node 直接运行；test/batteries/readme-examples.test.ts 会真跑它）：

import assert from 'node:assert/strict'
import { createMessages, memoryMessageStore } from '@weifuwujs/messages'

const messages = createMessages({ store: memoryMessageStore() })
await messages.send({ userId: 'user-1', kind: 'welcome', title: '欢迎使用', body: '从这条消息开始。', idempotencyKey: 'welcome:user-1' })
await messages.send({ userId: 'user-1', kind: 'welcome', title: '欢迎使用', body: '从这条消息开始。', idempotencyKey: 'welcome:user-1' })

const page = await messages.list({ userId: 'user-1' })
assert.equal(page.items.length, 1)                        // 幂等键：重复发送只入库一次
assert.equal(await messages.unread('user-1'), 1)

示例标注：ts exec = 由测试真跑；ts fragment = 片段（依赖前后文，不独立运行）；自足示例带包标签（如 ts tasks），由测试编译。

send 是服务端 API，不暴露 HTTP（通知面不能被客户端伪造）；消息落库后在线的本人立刻收到 WS 推送。

配合 auth 的注册欢迎信 recipe：

import { createApp } from '@weifuwujs/weifuwu'
import { createAuth } from '@weifuwujs/auth'
import { sqliteAuthStore } from '@weifuwujs/auth/sqlite'
import { createMessages } from '@weifuwujs/messages'
import { sqliteMessageStore } from '@weifuwujs/messages/sqlite'

const app = createApp()
const messages = createMessages({ store: sqliteMessageStore('data/messages.db') })
const auth = createAuth({
  store: sqliteAuthStore('data/auth.db'),
  hooks: {
    onRegister: (user) => {
      void messages.send({
        userId: user.id,
        kind: 'welcome',
        title: '欢迎使用',
        idempotencyKey: `welcome:${user.id}`, // 重投/重试只入一条
      })
    },
  },
})

auth.install(app) // HTTP + WS 统一鉴权（ctx.auth.user）
app.mount(auth)
messages.install(app) // 必须叠在 auth 之后：REST/WS 都看 ctx.auth.user

后台任务完成通知（tasks recipe）

长任务在 ./tasks 里跑完，用消息把结果送到用户手里；idempotencyKey 按业务主键命名， 任务重试（至少一次执行）也不会重复打扰用户：

import { createApp } from '@weifuwujs/weifuwu'
import { createMessages, memoryMessageStore } from '@weifuwujs/messages'
import { createTasks, memoryTaskStore } from '@weifuwujs/tasks'

const messages = createMessages({ store: memoryMessageStore() })
const tasks = createTasks({ store: memoryTaskStore() })

tasks.handle(
  'generate-report',
  async (payload: { userId: string; reportId: string }) => {
    await messages.send({
      userId: payload.userId,
      kind: 'report',
      title: '报表已生成',
      idempotencyKey: `report-done:${payload.reportId}`, // 任务重试不重复通知
    })
  },
)

const app = createApp()
messages.install(app)
tasks.install(app) // worker 随 app.close() 优雅停止

await tasks.enqueue(
  'generate-report',
  { userId: 'user-1', reportId: 'r1' },
  { idempotencyKey: 'report:r1' },
)

API 速览

createMessages(options)

| 选项 | 默认 | 说明 | | --- | --- | --- | | store | 必填 | 三驱动之一或你自己的 MessageStore | | bus | 进程内 | localMessageBus()；多副本换 redisMessageBus()（见下） | | onSend | — | 新消息落库后回调；抛错只告警（已持久化），幂等命中不触发 | | routePrefix | /messages | REST 前缀 | | pageSize | 20（上限 100） | 列表默认页大小 | | syncLimit | 100 | 断线补投单次最大条数（hasMore 后续页） | | now / logger | — | 可注入时钟、日志器 |

messages 实例：send / list / unread / markRead / markAllRead / routes / install / connections / publish / ready / close / migrate。

send(input) → { message, created }：input = { userId, title, kind?, body?, data?, idempotencyKey? }； created: false 表示幂等键命中（返回已有消息，不覆盖）。推荐键名 <动作>:<业务主键>。

REST（messages.install(app) / app.mount(messages) 后；一律 401 守卫 + 仅本人）：

| 方法 | 路径 | 说明 | | --- | --- | --- | | GET | /messages | 本人消息（unread/kind/limit/cursor 查询）→ { items, nextCursor, unread }（按 (createdAt,id) 降序） | | GET | /messages/unread | { unread }（徽标轮询兜底） | | POST | /messages/read | { ids } → { updated }（幂等） | | POST | /messages/read-all | → { updated } |

WebSocket 协议

连接 GET /messages/ws（未认证 401）；?since=<cursor> 控制补投：

• 连接建立后先收到 { type: 'unread', unread } 快照，再收实时事件；
• since 非空且合法 → 先推 { type: 'sync', items, hasMore, unread }（升序，subscribe-first： 宁可重投不可丢失）；hasMore 时用最后一条的游标再连一次续补；
• since 为空/省略 → 视为全量 sync；非法游标 → 1008 关闭；
• 之后：send → { type: 'message', message } + { type: 'unread', unread }； markRead/markAllRead → { type: 'unread', unread }。

投递语义是至少一次：断线重连可能重投，客户端按 message.id 去重即恰好一次。unread 是快照值， 不是增量。

断线不丢不重（客户端重连配方）

客户端记住"最新已见消息"的游标（列表升序取 items[0]，或 sync 的最后一条），重连时带上：

import { encodeCursor } from '@weifuwujs/messages'

// 收到列表/同步结果后：
const latest = page.items[0] // 降序列表的第一条 = 最新
const since = latest
  ? encodeCursor({ id: latest.id, createdAt: new Date(latest.createdAt) })
  : ''

// 重连：ws://…/messages/ws?since=<since>，把 sync.items + 实时 message 合并进同一个 Map
const byId = new Map<string, { id: string }>()
for (const item of [...sync.items, ...live]) byId.set(item.id, item)

重复应用同一批事件时 byId 不变——这就是"不重复"的全部代价。

三驱动与运维

| 驱动 | 入口 | 建表 | 说明 | | --- | --- | --- | --- | | memory | @weifuwujs/messages 的 memoryMessageStore() | — | 测试/单进程 demo | | sqlite | @weifuwujs/messages/sqlite | 构造即迁移 | 默认 weifuwu-messages migrate 目标（sqlite:data/messages.db） | | postgres | @weifuwujs/messages/postgres | 构造即迁移 + advisory lock | 多副本必须：UNIQUE(user_id, idempotency_key) 让幂等原子 |

weifuwu-messages migrate                                    # sqlite:data/messages.db
weifuwu-messages migrate --store postgres://user:pw@host/db # 多进程并发安全（pg advisory xact lock）

redis 不作为消息主存（不保证持久，丢了就是丢了）；它的位置在下面的实时总线。

保留与清理（1.3）

站内信表会无限增长。清理默认关闭——不配置策略 = 永不删除；开启后由显式清理触发 （cron/运维命令），不是后台自动任务：

const messages = createMessages({ store, retain: { maxAgeMs: 30 * 24 * 3600_000, maxPerUser: 5_000 } })

await messages.purge({ dryRun: true }) // { matched, deleted: 0, dryRun: true } —— 只报数，零删除
await messages.purge()                 // { matched, deleted, dryRun: false } —— 幂等：二次执行为 0

weifuwu-messages purge --max-age-ms 2592000000 --dry-run   # stdout 纯 JSON，可 | jq
weifuwu-messages purge --max-per-user 5000 --store postgres://user:pw@host/db

CLI 惯例（1.3 起五包统一，见 docs/alignment.md）：无命令 / 未知命令 → USAGE exit 2 且不写 stdout； --help → stdout + exit 0；退出码 0 成功 / 1 运行时错误 / 2 用法错误。

语义（三条都要读）：

1. maxAgeMs 是全局 TTL：删除 createdAt 早于 now - maxAgeMs 的消息（对所有用户生效）；
2. maxPerUser 按用户独立：每个用户只保留最近 N 条，不会因为别人多而删你的；
3. 未读不豁免：删除只看时间/条数，未读消息也会被删——这是显式开启保留策略的结果， 不是静默行为。需要保底可见性时把 maxAgeMs 设得比业务兜底周期更长，或用 maxPerUser 而非 TTL。

实现约束：olderThan 必须早于当前时刻（防误删全库，调用会显式报错）；清理走驱动原生 SQL （sqlite/postgres 各一条窗口函数删除），memory/sqlite/postgres 三驱动同一套契约测试。

多副本部署

单进程默认 localMessageBus()（进程内 fan-out）。多副本时每个实例都连同一个主存，并把总线换成 Redis：

import { createMessages } from '@weifuwujs/messages'
import { postgresMessageStore } from '@weifuwujs/messages/postgres'
import { redisMessageBus } from '@weifuwujs/messages/redis'

const bus = redisMessageBus('redis://localhost:6379', { prefix: 'myapp:messages' })
const messages = createMessages({
  store: postgresMessageStore('postgres://user:pw@localhost:5432/app'),
  bus,
})

await messages.ready() // 启动时等一次订阅就绪；之后 send 先发本节点再广播
// … 服务运行：A 实例 send，持有 WS 的 B 实例也会推送
await messages.close() // 只关自有连接与自有 bus；注入的 bus 生命周期归调用方
await bus.close() // 所以注入时由你负责关闭

• 按用户按需订阅 prefix:<userId>：本节点有该用户连接才订阅，最后一个连接断开后自动退订；
• redisMessageBus(url) 自己创建/关闭连接；传已有 ioredis 客户端则只 duplicate() 订阅连接，不关注入的客户端；
• 跨节点仍是至少一次：WS 下多节点可能对同一用户重复投递，客户端按 id 去重。

自定义驱动

实现 MessageStore（insert/list/unreadCount/markRead/markAllRead/migrate?）后，先用同一套 契约验证再上线：

import { memoryMessageStore } from '@weifuwujs/messages'
import { messagesStoreContract } from '@weifuwujs/messages/testing'

// 换成你的驱动：messagesStoreContract('my-store', () => myMessageStore())
messagesStoreContract('my-store', () => memoryMessageStore())

契约覆盖：幂等键不覆盖、降序分页稳定（同毫秒按 id）、只影响本人、markRead 幂等、未读计数。

测试辅助（/testing）

自定义 MessageStore / MessageBus 驱动与内置驱动共跑同一套契约：

import {
  messagesStoreContract,
  messageBusContract,
} from '@weifuwujs/messages/testing'

messagesStoreContract('my-store', () => myMessageStore())
messageBusContract('my-bus', () => myMessageBus())
// 第三个参数可选：{ skip: true } 或 { skip: '未实现清理' }

工厂会被套件多次调用：每次须返回独立的 store（sqlite 驱动用 ':memory:' 或唯一路径）—— 复用同一文件会状态泄漏并假红。

契约覆盖：insert/get（IDOR 防线）、幂等键、游标分页不重不漏、unread 计数与已读隔离、 listAfter 补投、watch/unwatch 订阅泄漏回归。

安全与边界

• 所有 store 查询强制 userId，越权读取/标记直接表现为"查无此件"；REST/WS 未认证 401；
• title/body 按纯文本存储，渲染层负责转义；data 只透传、不参与查询；
• app.close() 会断开全部 WS；messages.close() 幂等，注入 bus 时先关 messages 再关 bus；
• 群发/广播是多次 send（没有 fan-out API），量级大时用 ./tasks 异步扇出；
• 明确非目标：IM、跨服务消息总线、消息编辑/撤回、附件。