From d6f6a92bab146952715d0ae0a081fbe67730d1af Mon Sep 17 00:00:00 2001 From: kyugao Date: Fri, 7 Aug 2026 13:36:56 +0800 Subject: [PATCH] Initialize LineUp app server --- .env.example | 3 + .gitignore | 5 + DESIGN.md | 333 +++++++++++++++++++++++++ assets/chat/index.html | 204 +++++++++++++++ assets/chat/interaction-kernel.js | 75 ++++++ assets/chat/interaction-kernel.test.js | 29 +++ configs/lineup.yaml | 42 ++++ go.mod | 50 ++++ go.sum | 121 +++++++++ internal/hub.go | 60 +++++ internal/wkapi/wkapi.go | 214 ++++++++++++++++ internal/wkapi/wkapi_test.go | 69 +++++ main.go | 90 +++++++ modules/agent.go | 96 +++++++ modules/agent_test.go | 47 ++++ modules/channel.go | 75 ++++++ modules/client.go | 88 +++++++ modules/message.go | 143 +++++++++++ modules/modules.go | 58 +++++ modules/user.go | 95 +++++++ modules/webhook.go | 151 +++++++++++ package-lock.json | 67 +++++ package.json | 18 ++ run-server.sh | 51 ++++ 24 files changed, 2184 insertions(+) create mode 100644 .env.example create mode 100644 .gitignore create mode 100644 DESIGN.md create mode 100644 assets/chat/index.html create mode 100644 assets/chat/interaction-kernel.js create mode 100644 assets/chat/interaction-kernel.test.js create mode 100644 configs/lineup.yaml create mode 100644 go.mod create mode 100644 go.sum create mode 100644 internal/hub.go create mode 100644 internal/wkapi/wkapi.go create mode 100644 internal/wkapi/wkapi_test.go create mode 100644 main.go create mode 100644 modules/agent.go create mode 100644 modules/agent_test.go create mode 100644 modules/channel.go create mode 100644 modules/client.go create mode 100644 modules/message.go create mode 100644 modules/modules.go create mode 100644 modules/user.go create mode 100644 modules/webhook.go create mode 100644 package-lock.json create mode 100644 package.json create mode 100755 run-server.sh diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..f8d3950 --- /dev/null +++ b/.env.example @@ -0,0 +1,3 @@ +# Private deployment configuration for run-server.sh. +# Copy to .env, generate a strong random value, and never commit it. +LINEUP_AGENT_SHARED_SECRET= diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..57042ce --- /dev/null +++ b/.gitignore @@ -0,0 +1,5 @@ +.env +node_modules/ +bin/ +*.log +*.pid diff --git a/DESIGN.md b/DESIGN.md new file mode 100644 index 0000000..43e0f96 --- /dev/null +++ b/DESIGN.md @@ -0,0 +1,333 @@ +# LineUp App Server — 项目设计文档 + +> 版本：0.1（提案） +> 日期：2026-07-30 +> 状态：待确认 +> 目标：精简的蓝牙应用服务端，帮助用户与远端 Agent 服务建立连接，实现互相通信 + +--- + +## 1. 项目定位 + +### 1.1 与唐僧叨叨的关系 + +唐僧叨叨是一个完整的 IM 业务系统（用户/群组/文件/朋友圈/工作台/推送...），模块众多、功能庞大。LineUp App Server 不是唐僧叨叨的副本，而是： + +``` +唐僧叨叨：消费级 IM 全栈（群聊、朋友圈、文件、搜索...） +LineUp App Server：只做两件事 —— + ① 用户认证身份 → 接入通讯网络 + ② 消息路由 → 连接用户与 Agent +``` + +**所以我们不是在唐僧叨叨上改，而是从它那里取经：** +- 复用它的 WuKongIM 集成模式（怎么调用 API、怎么写 Webhook） +- 复用它的 HTTP 路由注册方式（gin 框架 + module 自注册） +- 复用它的 Webhook 消息处理逻辑（msg.notify、msg.offline 等事件） +- 丢弃所有跟业务无关的模块（群聊、文件、推送、朋友圈...） + +### 1.2 LineUp App Server 的定义 + +> LineUp App Server 是一个轻量级的**通讯网关**。 +> 它的唯一职责是：让一个通过蓝牙/WiFi 连接的移动客户端，能够接入 WuKongIM 3.0 通讯网络，并与远端的 Agent 服务进行消息交换。 + +**它不是什么：** +- 不是社交平台（不处理好友、朋友圈、群聊） +- 不是文件存储（不处理文件上传/下载） +- 不是推送网关（不处理 APNS/HMS/Mi Push） +- 不是管理后台（不处理统计、审核、工作台） +- 不是机器人平台（不处理自定义机器人接入） + +--- + +## 2. 核心概念 + +### 2.1 三种参与者 + +``` +┌──────────┐ ┌─────────────────┐ ┌──────────┐ +│ 用户 Client │ ◄─► │ LineUp App Server │ ◄─► │ Agent 服务 │ +│ (手机 App) │ │ (通讯网关) │ │ (远端 LLM) │ +└──────────┘ └────────┬────────┘ └──────────┘ + │ + ┌──────▼──────┐ + │ WuKongIM 3.0 │ + │ (消息管道) │ + └──────────────┘ +``` + +- **用户 Client**：运行在手机上的 App，通过蓝牙/WiFi 与 App Server 通信 +- **Agent 服务**：远端的 AI Agent（如 Hermes），有自己的 WuKongIM 客户端连接 +- **WuKongIM 3.0**：消息传输管道，负责可靠投递 +- **LineUp App Server**：通讯网关，负责身份认证 + 消息路由 + +### 2.2 消息流转模型 + +``` +用户发送消息 "帮我查天气" + + ① Client → App Server + 用户 App 通过 HTTP/WS 发送消息到 App Server + + ② App Server → WuKongIM + App Server 调用 WuKongIM API 将消息投递到 Agent 所在的频道 + + ③ WuKongIM → Agent Service + WuKongIM 将消息推送到 Agent 的 WebSocket 连接 + + ④ Agent Service 处理 + Agent 处理请求（可能调用外部工具） + + ⑤ Agent Service → WuKongIM + Agent 回复消息，发送到 WuKongIM + + ⑥ WuKongIM → App Server (Webhook) + WuKongIM 通过 Webhook 通知 App Server 有新消息 + + ⑦ App Server → Client + App Server 将消息推送给用户 Client +``` + +--- + +## 3. 通信协议分层 + +``` +┌─────────────────────────────────────────┐ +│ LineUp 自定义消息协议 │ ← 业务层 +│ (text, choice, progress, canvas...) │ +├─────────────────────────────────────────┤ +│ 唐僧叨叨消息模型 │ ← 适配层 +│ (Message, Channel, Conversation) │ +├──────────────────┬──────────────────────┤ +│ WKProto WebSocket│ HTTP API + Webhook │ ← WuKongIM 3.0 +│ (双向实时, 15200) │ (管理面, 15001) │ +│ │ (通知面, callback) │ +└──────────────────┴──────────────────────┘ +``` + +**WuKongIM 3.0 的三层能力:** + +| 层 | 协议 | 端口 | 方向 | 用途 | +|---|---|---|---|---| +| **数据面** | WKProto WebSocket / TCP | 15200 / 15100 | **双向全双工** | 客户端收发消息 | +| **管理面** | HTTP REST API | 15001 | App Server → WK | 用户/频道/消息管理 | +| **通知面** | HTTP Webhook | Callback | WK → App Server | 事件通知(新消息/离线/在线) | + +> 客户端直连 WuKongIM 的 WebSocket (15200),所有实时消息都在这个管道里跑。 +> App Server 只做两件事: ① 通过 HTTP API 做管理; ② 通过 Webhook 接收事件推送给 Client。 + +--- + +## 4. 架构设计 + +### 4.1 服务端口 + +``` +LineUp App Server: + - HTTP API: :8090 (客户端 REST API) + - WebSocket: :8090/ws (客户端实时推送) + - Webhook: :8090/v1/webhook (接收 WuKongIM 回调) + +WuKongIM 3.0: (已部署) + - API: :15001 (管理 API: /route, /channel, /message/send...) + - TCP: :15100 (WKProto TCP) + - WebSocket: :15200 (WKProto WebSocket) + - Manager: :15301 (管理后台) +``` + +### 4.2 模块划分 + +``` +lineup-app-server/ +├── main.go # 入口：加载配置 → 注册模块 → 启动 HTTP +├── go.mod / go.sum +├── configs/ +│ └── lineup.yaml # 配置：WuKongIM 地址、Token、端口 +├── internal/ +│ ├── app.go # App 生命周期：Init → Start → Stop +│ ├── config.go # 配置结构体 +│ └── db.go # 数据库（轻量，仅存储用户/Token） +└── modules/ + ├── user/ # 用户模块：登录/注册/Token管理 + │ ├── module.go # → 模块自注册 + │ ├── api.go # → REST: POST /login, POST /register + │ ├── db.go # → 用户数据存储 + │ └── service.go # → 业务逻辑 + ├── channel/ # 频道模块：创建Agent对话频道 + │ ├── module.go + │ ├── api.go # → REST: POST /channels, GET /channels + │ └── service.go # → 调用 WuKongIM Channel API + ├── message/ # 消息模块：收发消息 + │ ├── module.go + │ ├── api.go # → REST: POST /messages, GET /messages/sync + │ └── service.go # → 调用 WuKongIM Message API + ├── webhook/ # Webhook模块：接收WuKongIM回调 + │ ├── module.go + │ ├── handler.go # → HTTP: POST /v1/webhook + │ └── event.go # → 事件分发：msg.notify → 推送Client + └── client/ # 客户端连接模块（蓝牙/WS网关） + ├── module.go + ├── ws.go # → WebSocket: /ws (向Client实时推送) + └── push.go # → 消息推送：把WuKongIM消息推给客户端 +``` + +**只有 6 个模块（vs 唐僧叨叨的 15 个），代码量预计减少 70%+。** + +### 4.3 数据库设计（最简方案） + +选项 A（推荐起步）: **不使用关系数据库，用文件/内存存储** +```go +// 用户注册本质上就是向WuKongIM申请一个token +// agent的对话频道和消息都存储在WuKongIM内部 +// App Server 本身不需要持久化存储 +``` + +选项 B（如果需要持久化）: **SQLite / PebbleDB** +```sql +-- 极简表设计 +CREATE TABLE users ( + uid TEXT PRIMARY KEY, -- 用户ID（与WuKongIM一致） + token TEXT, -- WuKongIM Token + name TEXT, -- 显示名称 + created INTEGER -- 创建时间 +); +``` + +--- + +## 5. 核心通讯流程 + +### 5.1 用户上线流程 + +``` +Step 1: Client 通过 HTTP POST /login + → App Server 验证用户凭证 + → App Server 调用 WuKongIM API 获取 Route 信息 + → 返回 {token, route} 给 Client + +Step 2: Client 使用 token 和 route 直连 WuKongIM WebSocket + → WuKongIM WebSocket: ws://host:15200?token=xxx + → 这一步完全绕开 App Server，Client 自己跟 WuKongIM 通信 + +Step 3: Client 同时连接 App Server WebSocket + → ws://app-server:8090/ws?token=xxx + → 用于接收实时推送（Webhook 转发过来的消息） +``` + +**关键设计决策：客户端直连 WuKongIM，App Server 只作为 Webhook 中转。** + +### 5.2 发送消息流程 + +``` +Client → WuKongIM WebSocket (直连) + → 消息直接写入 WuKongIM，不经过 App Server + → App Server 通过 Webhook 收到通知后，可以通过 WebSocket 推送给对端 +``` + +### 5.3 Agent 连接方式 + +``` +Agent Service 有两种接入方式： + +方式 A（推荐）：直接作为 WuKongIM 客户端 + - Agent 用 Go/Python SDK 连接 WuKongIM + - 收发消息走 WuKongIM 原生协议 + - App Server 完全透明，只做 Webhook 中转 + +方式 B：通过 App Server API + - Agent 注册为 App Server 的"特殊用户" + - 消息通过 App Server 的 HTTP API 中转 + - 适合 Agent 无法直连 WuKongIM 的场景 +``` + +--- + +## 6. WuKongIM 3.0 集成清单 + +### 6.1 调用的 WuKongIM API + +| API | 用途 | 代码位置 | +|---|---|---| +| `POST /user/token` | 生成用户 Token | `modules/user/service.go` | +| `GET /route?uid=` | 获取连接地址 | `modules/user/service.go` | +| `POST /channel` | 创建 Agent 对话频道 | `modules/channel/service.go` | +| `POST /channel/messagesync` | 同步频道消息 | `modules/message/service.go` | +| `POST /message/send` | 发送消息（App Server 代发） | `modules/message/service.go` | + +### 6.2 接收的 Webhook 事件 + +| 事件 | 含义 | 处理方式 | +|---|---|---| +| `msg.notify` | 新消息通知 | 推送给对应的 Client WebSocket | +| `msg.offline` | 离线消息 | 推送给刚上线的 Client | +| `user.onlinestatus` | 用户在线状态变化 | 更新 Agent 在线状态 | + +### 6.3 v3 兼容性对照（已有验证结论） + +| 功能 | 状态 | 备注 | +|---|---|---| +| Token 生成 | ✅ | API 兼容 | +| Route 查询 | ✅ | 格式兼容（uid参数在v3被忽略但不影响） | +| Channel 管理 | ✅ | API 兼容 | +| Message Send | ✅ | API 兼容 | +| Message Sync | ✅ | v3 使用 `channel/messagesync` | +| Webhook 事件名 | ✅ | msg.notify/msg.offline/user.onlinestatus 完全一致 | +| Webhook HTTP | ✅ | v3 原生 HTTP Webhook（不需要改 gRPC 适配） | +| Message Search | ❌ 不需要 | 精简版不做消息搜索 | +| 文件上传 | ❌ 不需要 | 精简版不做文件服务 | + +--- + +## 7. 实施计划 + +### Phase 1: 最小骨架（1-2 天） +``` +目标：Go 项目编译通过，HTTP 服务启动，能连接 WuKongIM 3.0 + + □ 创建 go.mod，引入唐僧叨叨的 server/config 和 wkhttp + □ 实现 main.go：加载配置 + 启动 gin HTTP 服务 + □ 实现 user 模块：POST /login（调 WuKongIM /user/token） + □ 实现 webhook 模块：POST /v1/webhook（接收并打印事件日志） + □ 手工测试：curl /login → 拿到 token → 用 token 连 WuKongIM +``` + +### Phase 2: 消息通路（2-3 天） +``` +目标：用户和 Agent 能通过 App Server + WuKongIM 互相发消息 + + □ 实现 channel 模块：创建/查询频道 + □ 实现 message 模块：发送消息、同步消息 + □ 实现 client WebSocket：向 Client 实时推送消息 + □ 端到端测试：用户发消息 → Agent 收到 → Agent 回复 → 用户收到 +``` + +### Phase 3: 蓝牙/本地通信适配（2-3 天） +``` +目标：手机 App 通过本地网络连接 App Server + + □ 实现蓝牙发现 / 局域网发现协议 + □ App Server 暴露本地 HTTP 服务（mDNS 广播） + □ 安全认证：设备配对 Token +``` + +### Phase 4: 稳定与优化（1-2 天） +``` + □ 连接断开重连 + □ 消息确认与去重 + □ 监控与日志 + □ 性能调优 +``` + +--- + +## 8. 参考资料 + +- WuKongIM 3.0 源码：`wukongim/` +- WuKongIM 3.0 本地环境：`infra/wukongim-v3/` +- 兼容性验证：`项目状态记录.md` §六-§八 +- LineUp App 层架构方案：`设计/02.正式方案/lineup-app-layer-architecture.md` + +--- + +*本文档为项目设计提案，待确认后进入 Phase 1 实现。* diff --git a/assets/chat/index.html b/assets/chat/index.html new file mode 100644 index 0000000..399ab4f --- /dev/null +++ b/assets/chat/index.html @@ -0,0 +1,204 @@ + + + + + + + LineUp + + + + + + +
+

LineUp

你的 AI 搭档

+ + + +
+
+
+
+

AI Agent

正在连接…
+ +
+
+
+
+ + + diff --git a/assets/chat/interaction-kernel.js b/assets/chat/interaction-kernel.js new file mode 100644 index 0000000..7176206 --- /dev/null +++ b/assets/chat/interaction-kernel.js @@ -0,0 +1,75 @@ +/* + * Platform-neutral LineUp presentation kernel. + * + * It converts untrusted transport payloads into a small, validated set of + * ConversationItems. UI hosts register trusted renderers for those items; + * neither a renderer nor a UI Surface needs to parse IM payloads directly. + */ +(function exposeLineUpKernel(root, factory) { + const api = factory(); + if (typeof module === 'object' && module.exports) module.exports = api; + root.LineUp = Object.assign(root.LineUp || {}, api); +})(typeof globalThis === 'undefined' ? window : globalThis, function createKernel() { + const PROTOCOL_VERSION = 1; + const MAX_PAYLOAD_BYTES = 192 * 1024; + const UI_TYPES = new Set(['lineup.v1.ui.open', 'lineup.v1.ui.patch', 'lineup.v1.ui.close']); + + function isObject(value) { return value !== null && typeof value === 'object' && !Array.isArray(value); } + function byteLength(value) { return new TextEncoder().encode(String(value)).length; } + function text(value) { return typeof value === 'string' ? value : ''; } + function item(kind, data, envelope) { + return Object.freeze({ kind, data: Object.freeze(data), envelope: envelope || null, receivedAt: Date.now() }); + } + + class ConversationItemFactory { + decode(rawPayload) { + const raw = String(rawPayload ?? ''); + if (byteLength(raw) > MAX_PAYLOAD_BYTES) return item('protocol-fallback', { reason: 'payload_too_large' }); + let value; + try { value = JSON.parse(raw); } catch { return item('markdown', { markdown: raw, source: 'plain' }); } + if (!isObject(value) || typeof value.type !== 'string') return item('markdown', { markdown: raw, source: 'plain' }); + if (value.v !== PROTOCOL_VERSION || !value.type.startsWith('lineup.v1.')) return this.legacy(value); + return this.envelope(value); + } + + envelope(envelope) { + const payload = isObject(envelope.payload) ? envelope.payload : {}; + switch (envelope.type) { + case 'lineup.v1.text': + return item('markdown', { markdown: text(payload.text), source: 'agent' }, envelope); + case 'lineup.v1.agent.status': + return item('agent-status', { status: text(payload.status) || 'unknown', detail: text(payload.detail) }, envelope); + case 'lineup.v1.agent.progress': + return item('progress', { title: text(payload.title) || '处理中', percent: Number(payload.percent) || 0, status: text(payload.status) }, envelope); + case 'lineup.v1.error': + return item('agent-error', { code: text(payload.code), message: text(payload.message) || '未知错误' }, envelope); + case 'lineup.v1.app.call': + return item('app-call', payload, envelope); + default: + if (UI_TYPES.has(envelope.type)) return item(envelope.type.slice('lineup.v1.'.length), payload, envelope); + return item('protocol-fallback', { reason: 'unsupported_type', type: envelope.type }, envelope); + } + } + + legacy(value) { + if (value.v === undefined && typeof value.type === 'string') return item('legacy-card', value); + return item('protocol-fallback', { reason: 'unsupported_version', version: value.v, type: value.type }); + } + } + + class RendererRegistry { + constructor() { this.renderers = new Map(); } + register(kind, renderer) { + if (typeof kind !== 'string' || typeof renderer !== 'function') throw new TypeError('Renderer registration requires a kind and a function'); + this.renderers.set(kind, renderer); + return this; + } + render(conversationItem, context) { + const renderer = this.renderers.get(conversationItem.kind) || this.renderers.get('protocol-fallback'); + if (!renderer) throw new Error(`No renderer registered for ${conversationItem.kind}`); + return renderer(conversationItem, context); + } + } + + return { PROTOCOL_VERSION, MAX_PAYLOAD_BYTES, ConversationItemFactory, RendererRegistry, isObject }; +}); diff --git a/assets/chat/interaction-kernel.test.js b/assets/chat/interaction-kernel.test.js new file mode 100644 index 0000000..7b51df8 --- /dev/null +++ b/assets/chat/interaction-kernel.test.js @@ -0,0 +1,29 @@ +const test = require('node:test'); +const assert = require('node:assert/strict'); +const { ConversationItemFactory, RendererRegistry } = require('./interaction-kernel.js'); + +test('decodes plain agent output as a markdown conversation item', () => { + const item = new ConversationItemFactory().decode('# 标题\n\n- 一项'); + assert.equal(item.kind, 'markdown'); + assert.equal(item.data.markdown, '# 标题\n\n- 一项'); +}); + +test('decodes a v1 surface lifecycle message without treating it as a card', () => { + const item = new ConversationItemFactory().decode(JSON.stringify({ + v: 1, type: 'lineup.v1.ui.open', conversation_id: 'conv-1', + payload: { instance_id: 'task.01', app: { id: 'com.lineup.task', version: '1.0.0' }, state: { phase: 'running' } }, + })); + assert.equal(item.kind, 'ui.open'); + assert.equal(item.data.instance_id, 'task.01'); +}); + +test('uses a safe fallback for an unsupported protocol message', () => { + const item = new ConversationItemFactory().decode(JSON.stringify({ v: 1, type: 'lineup.v1.secret.execute', payload: {} })); + assert.equal(item.kind, 'protocol-fallback'); + assert.equal(item.data.reason, 'unsupported_type'); +}); + +test('routes only validated conversation items through registered renderers', () => { + const registry = new RendererRegistry().register('markdown', item => item.data.markdown).register('protocol-fallback', () => 'fallback'); + assert.equal(registry.render(new ConversationItemFactory().decode('hello')), 'hello'); +}); diff --git a/configs/lineup.yaml b/configs/lineup.yaml new file mode 100644 index 0000000..1281cdb --- /dev/null +++ b/configs/lineup.yaml @@ -0,0 +1,42 @@ +# LineUp App Server 配置 +mode: "debug" +addr: ":8090" + +# Trusted cross-origin clients. Browser chat is served same-origin; Tauri's +# WebView origin must be explicit rather than allowing arbitrary web pages. +client: + allowedOrigins: + - "tauri://localhost" + - "https://tauri.localhost" + - "http://tauri.localhost" + # Tauri `dev` loads the trusted Host UI from this fixed local Vite URL. + # Keep it explicit; do not replace this allowlist with a wildcard. + - "http://127.0.0.1:1420" + - "http://localhost:1420" + # Browser preview / fallback ports used when the default dev port is occupied. + - "http://127.0.0.1:1421" + - "http://127.0.0.1:1422" + - "http://localhost:1421" + - "http://localhost:1422" + # Linux development Host, accessed by the Mac browser through Tailscale. + # Keep the exact origin; never replace this allowlist with a wildcard. + - "http://100.121.118.116:1420" + # Same trusted development Host, accessed by phones on the local Wi-Fi. + # Keep this exact LAN origin; never replace this allowlist with a wildcard. + - "http://192.168.3.108:1420" + +# WuKongIM 3.0 连接配置 +wukongim: + apiURL: "http://127.0.0.1:15001" + wsAddr: "ws://127.0.0.1:15200" + managerToken: "lineup-dev-only" + +# 默认 Agent 配置 +agent: + # The stable identity used by the real Hermes Adapter. + uid: "agent_hermes_main" + channelID: "agent_default_channel" + channelType: 2 + # Set via LINEUP_AGENT_SHARED_SECRET in the deployment environment. Do not + # store its value in this repository; heartbeat is disabled until configured. + shared_secret: "" diff --git a/go.mod b/go.mod new file mode 100644 index 0000000..5aba4c0 --- /dev/null +++ b/go.mod @@ -0,0 +1,50 @@ +module github.com/lineup/app-server + +go 1.25.11 + +require ( + github.com/gin-gonic/gin v1.12.0 + github.com/spf13/viper v1.21.0 +) + +require ( + github.com/bytedance/gopkg v0.1.3 // indirect + github.com/bytedance/sonic v1.15.0 // indirect + github.com/bytedance/sonic/loader v0.5.0 // indirect + github.com/cloudwego/base64x v0.1.6 // indirect + github.com/fsnotify/fsnotify v1.9.0 // indirect + github.com/gabriel-vasile/mimetype v1.4.12 // indirect + github.com/gin-contrib/sse v1.1.0 // indirect + github.com/go-playground/locales v0.14.1 // indirect + github.com/go-playground/universal-translator v0.18.1 // indirect + github.com/go-playground/validator/v10 v10.30.1 // indirect + github.com/go-viper/mapstructure/v2 v2.4.0 // indirect + github.com/goccy/go-json v0.10.5 // indirect + github.com/goccy/go-yaml v1.19.2 // indirect + github.com/gorilla/websocket v1.5.3 // indirect + github.com/json-iterator/go v1.1.12 // indirect + github.com/klauspost/cpuid/v2 v2.3.0 // indirect + github.com/leodido/go-urn v1.4.0 // indirect + github.com/mattn/go-isatty v0.0.20 // indirect + github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect + github.com/modern-go/reflect2 v1.0.2 // indirect + github.com/pelletier/go-toml/v2 v2.2.4 // indirect + github.com/quic-go/qpack v0.6.0 // indirect + github.com/quic-go/quic-go v0.59.0 // indirect + github.com/sagikazarmark/locafero v0.11.0 // indirect + github.com/sourcegraph/conc v0.3.1-0.20240121214520-5f936abd7ae8 // indirect + github.com/spf13/afero v1.15.0 // indirect + github.com/spf13/cast v1.10.0 // indirect + github.com/spf13/pflag v1.0.10 // indirect + github.com/subosito/gotenv v1.6.0 // indirect + github.com/twitchyliquid64/golang-asm v0.15.1 // indirect + github.com/ugorji/go/codec v1.3.1 // indirect + go.mongodb.org/mongo-driver/v2 v2.5.0 // indirect + go.yaml.in/yaml/v3 v3.0.4 // indirect + golang.org/x/arch v0.22.0 // indirect + golang.org/x/crypto v0.48.0 // indirect + golang.org/x/net v0.51.0 // indirect + golang.org/x/sys v0.41.0 // indirect + golang.org/x/text v0.34.0 // indirect + google.golang.org/protobuf v1.36.10 // indirect +) diff --git a/go.sum b/go.sum new file mode 100644 index 0000000..d44500d --- /dev/null +++ b/go.sum @@ -0,0 +1,121 @@ +github.com/bytedance/gopkg v0.1.3 h1:TPBSwH8RsouGCBcMBktLt1AymVo2TVsBVCY4b6TnZ/M= +github.com/bytedance/gopkg v0.1.3/go.mod h1:576VvJ+eJgyCzdjS+c4+77QF3p7ubbtiKARP3TxducM= +github.com/bytedance/sonic v1.15.0 h1:/PXeWFaR5ElNcVE84U0dOHjiMHQOwNIx3K4ymzh/uSE= +github.com/bytedance/sonic v1.15.0/go.mod h1:tFkWrPz0/CUCLEF4ri4UkHekCIcdnkqXw9VduqpJh0k= +github.com/bytedance/sonic/loader v0.5.0 h1:gXH3KVnatgY7loH5/TkeVyXPfESoqSBSBEiDd5VjlgE= +github.com/bytedance/sonic/loader v0.5.0/go.mod h1:AR4NYCk5DdzZizZ5djGqQ92eEhCCcdf5x77udYiSJRo= +github.com/cloudwego/base64x v0.1.6 h1:t11wG9AECkCDk5fMSoxmufanudBtJ+/HemLstXDLI2M= +github.com/cloudwego/base64x v0.1.6/go.mod h1:OFcloc187FXDaYHvrNIjxSe8ncn0OOM8gEHfghB2IPU= +github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/frankban/quicktest v1.14.6 h1:7Xjx+VpznH+oBnejlPUj8oUpdxnVs4f8XU8WnHkI4W8= +github.com/frankban/quicktest v1.14.6/go.mod h1:4ptaffx2x8+WTWXmUCuVU6aPUX1/Mz7zb5vbUoiM6w0= +github.com/fsnotify/fsnotify v1.9.0 h1:2Ml+OJNzbYCTzsxtv8vKSFD9PbJjmhYF14k/jKC7S9k= +github.com/fsnotify/fsnotify v1.9.0/go.mod h1:8jBTzvmWwFyi3Pb8djgCCO5IBqzKJ/Jwo8TRcHyHii0= +github.com/gabriel-vasile/mimetype v1.4.12 h1:e9hWvmLYvtp846tLHam2o++qitpguFiYCKbn0w9jyqw= +github.com/gabriel-vasile/mimetype v1.4.12/go.mod h1:d+9Oxyo1wTzWdyVUPMmXFvp4F9tea18J8ufA774AB3s= +github.com/gin-contrib/sse v1.1.0 h1:n0w2GMuUpWDVp7qSpvze6fAu9iRxJY4Hmj6AmBOU05w= +github.com/gin-contrib/sse v1.1.0/go.mod h1:hxRZ5gVpWMT7Z0B0gSNYqqsSCNIJMjzvm6fqCz9vjwM= +github.com/gin-gonic/gin v1.12.0 h1:b3YAbrZtnf8N//yjKeU2+MQsh2mY5htkZidOM7O0wG8= +github.com/gin-gonic/gin v1.12.0/go.mod h1:VxccKfsSllpKshkBWgVgRniFFAzFb9csfngsqANjnLc= +github.com/go-playground/assert/v2 v2.2.0 h1:JvknZsQTYeFEAhQwI4qEt9cyV5ONwRHC+lYKSsYSR8s= +github.com/go-playground/assert/v2 v2.2.0/go.mod h1:VDjEfimB/XKnb+ZQfWdccd7VUvScMdVu0Titje2rxJ4= +github.com/go-playground/locales v0.14.1 h1:EWaQ/wswjilfKLTECiXz7Rh+3BjFhfDFKv/oXslEjJA= +github.com/go-playground/locales v0.14.1/go.mod h1:hxrqLVvrK65+Rwrd5Fc6F2O76J/NuW9t0sjnWqG1slY= +github.com/go-playground/universal-translator v0.18.1 h1:Bcnm0ZwsGyWbCzImXv+pAJnYK9S473LQFuzCbDbfSFY= +github.com/go-playground/universal-translator v0.18.1/go.mod h1:xekY+UJKNuX9WP91TpwSH2VMlDf28Uj24BCp08ZFTUY= +github.com/go-playground/validator/v10 v10.30.1 h1:f3zDSN/zOma+w6+1Wswgd9fLkdwy06ntQJp0BBvFG0w= +github.com/go-playground/validator/v10 v10.30.1/go.mod h1:oSuBIQzuJxL//3MelwSLD5hc2Tu889bF0Idm9Dg26cM= +github.com/go-viper/mapstructure/v2 v2.4.0 h1:EBsztssimR/CONLSZZ04E8qAkxNYq4Qp9LvH92wZUgs= +github.com/go-viper/mapstructure/v2 v2.4.0/go.mod h1:oJDH3BJKyqBA2TXFhDsKDGDTlndYOZ6rGS0BRZIxGhM= +github.com/goccy/go-json v0.10.5 h1:Fq85nIqj+gXn/S5ahsiTlK3TmC85qgirsdTP/+DeaC4= +github.com/goccy/go-json v0.10.5/go.mod h1:oq7eo15ShAhp70Anwd5lgX2pLfOS3QCiwU/PULtXL6M= +github.com/goccy/go-yaml v1.19.2 h1:PmFC1S6h8ljIz6gMRBopkjP1TVT7xuwrButHID66PoM= +github.com/goccy/go-yaml v1.19.2/go.mod h1:XBurs7gK8ATbW4ZPGKgcbrY1Br56PdM69F7LkFRi1kA= +github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= +github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= +github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg= +github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg= +github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= +github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM= +github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo= +github.com/klauspost/cpuid/v2 v2.3.0 h1:S4CRMLnYUhGeDFDqkGriYKdfoFlDnMtqTiI/sFzhA9Y= +github.com/klauspost/cpuid/v2 v2.3.0/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0= +github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= +github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= +github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= +github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= +github.com/leodido/go-urn v1.4.0 h1:WT9HwE9SGECu3lg4d/dIA+jxlljEa1/ffXKmRjqdmIQ= +github.com/leodido/go-urn v1.4.0/go.mod h1:bvxc+MVxLKB4z00jd1z+Dvzr47oO32F/QSNjSBOlFxI= +github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY= +github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y= +github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= +github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w8PVh93nsPXa1VrQ6jlwL5oN8l14QlcNfg= +github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= +github.com/modern-go/reflect2 v1.0.2 h1:xBagoLtFs94CBntxluKeaWgTMpvLxC4ur3nMaC9Gz0M= +github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk= +github.com/pelletier/go-toml/v2 v2.2.4 h1:mye9XuhQ6gvn5h28+VilKrrPoQVanw5PMw/TB0t5Ec4= +github.com/pelletier/go-toml/v2 v2.2.4/go.mod h1:2gIqNv+qfxSVS7cM2xJQKtLSTLUE9V8t9Stt+h56mCY= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/quic-go/qpack v0.6.0 h1:g7W+BMYynC1LbYLSqRt8PBg5Tgwxn214ZZR34VIOjz8= +github.com/quic-go/qpack v0.6.0/go.mod h1:lUpLKChi8njB4ty2bFLX2x4gzDqXwUpaO1DP9qMDZII= +github.com/quic-go/quic-go v0.59.0 h1:OLJkp1Mlm/aS7dpKgTc6cnpynnD2Xg7C1pwL6vy/SAw= +github.com/quic-go/quic-go v0.59.0/go.mod h1:upnsH4Ju1YkqpLXC305eW3yDZ4NfnNbmQRCMWS58IKU= +github.com/rogpeppe/go-internal v1.10.0 h1:TMyTOH3F/DB16zRVcYyreMH6GnZZrwQVAoYjRBZyWFQ= +github.com/rogpeppe/go-internal v1.10.0/go.mod h1:UQnix2H7Ngw/k4C5ijL5+65zddjncjaFoBhdsK/akog= +github.com/sagikazarmark/locafero v0.11.0 h1:1iurJgmM9G3PA/I+wWYIOw/5SyBtxapeHDcg+AAIFXc= +github.com/sagikazarmark/locafero v0.11.0/go.mod h1:nVIGvgyzw595SUSUE6tvCp3YYTeHs15MvlmU87WwIik= +github.com/sourcegraph/conc v0.3.1-0.20240121214520-5f936abd7ae8 h1:+jumHNA0Wrelhe64i8F6HNlS8pkoyMv5sreGx2Ry5Rw= +github.com/sourcegraph/conc v0.3.1-0.20240121214520-5f936abd7ae8/go.mod h1:3n1Cwaq1E1/1lhQhtRK2ts/ZwZEhjcQeJQ1RuC6Q/8U= +github.com/spf13/afero v1.15.0 h1:b/YBCLWAJdFWJTN9cLhiXXcD7mzKn9Dm86dNnfyQw1I= +github.com/spf13/afero v1.15.0/go.mod h1:NC2ByUVxtQs4b3sIUphxK0NioZnmxgyCrfzeuq8lxMg= +github.com/spf13/cast v1.10.0 h1:h2x0u2shc1QuLHfxi+cTJvs30+ZAHOGRic8uyGTDWxY= +github.com/spf13/cast v1.10.0/go.mod h1:jNfB8QC9IA6ZuY2ZjDp0KtFO2LZZlg4S/7bzP6qqeHo= +github.com/spf13/pflag v1.0.10 h1:4EBh2KAYBwaONj6b2Ye1GiHfwjqyROoF4RwYO+vPwFk= +github.com/spf13/pflag v1.0.10/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg= +github.com/spf13/viper v1.21.0 h1:x5S+0EU27Lbphp4UKm1C+1oQO+rKx36vfCoaVebLFSU= +github.com/spf13/viper v1.21.0/go.mod h1:P0lhsswPGWD/1lZJ9ny3fYnVqxiegrlNrEmgLjbTCAY= +github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw= +github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo= +github.com/stretchr/objx v0.5.2/go.mod h1:FRsXN1f5AsAjCGJKqEizvkpNtU+EGNCLh3NxZ/8L+MA= +github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= +github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= +github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU= +github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo= +github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= +github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= +github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= +github.com/subosito/gotenv v1.6.0 h1:9NlTDc1FTs4qu0DDq7AEtTPNw6SVm7uBMsUCUjABIf8= +github.com/subosito/gotenv v1.6.0/go.mod h1:Dk4QP5c2W3ibzajGcXpNraDfq2IrhjMIvMSWPKKo0FU= +github.com/twitchyliquid64/golang-asm v0.15.1 h1:SU5vSMR7hnwNxj24w34ZyCi/FmDZTkS4MhqMhdFk5YI= +github.com/twitchyliquid64/golang-asm v0.15.1/go.mod h1:a1lVb/DtPvCB8fslRZhAngC2+aY1QWCk3Cedj/Gdt08= +github.com/ugorji/go/codec v1.3.1 h1:waO7eEiFDwidsBN6agj1vJQ4AG7lh2yqXyOXqhgQuyY= +github.com/ugorji/go/codec v1.3.1/go.mod h1:pRBVtBSKl77K30Bv8R2P+cLSGaTtex6fsA2Wjqmfxj4= +go.mongodb.org/mongo-driver/v2 v2.5.0 h1:yXUhImUjjAInNcpTcAlPHiT7bIXhshCTL3jVBkF3xaE= +go.mongodb.org/mongo-driver/v2 v2.5.0/go.mod h1:yOI9kBsufol30iFsl1slpdq1I0eHPzybRWdyYUs8K/0= +go.uber.org/mock v0.6.0 h1:hyF9dfmbgIX5EfOdasqLsWD6xqpNZlXblLB/Dbnwv3Y= +go.uber.org/mock v0.6.0/go.mod h1:KiVJ4BqZJaMj4svdfmHM0AUx4NJYO8ZNpPnZn1Z+BBU= +go.yaml.in/yaml/v3 v3.0.4 h1:tfq32ie2Jv2UxXFdLJdh3jXuOzWiL1fo0bu/FbuKpbc= +go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg= +golang.org/x/arch v0.22.0 h1:c/Zle32i5ttqRXjdLyyHZESLD/bB90DCU1g9l/0YBDI= +golang.org/x/arch v0.22.0/go.mod h1:dNHoOeKiyja7GTvF9NJS1l3Z2yntpQNzgrjh1cU103A= +golang.org/x/crypto v0.48.0 h1:/VRzVqiRSggnhY7gNRxPauEQ5Drw9haKdM0jqfcCFts= +golang.org/x/crypto v0.48.0/go.mod h1:r0kV5h3qnFPlQnBSrULhlsRfryS2pmewsg+XfMgkVos= +golang.org/x/net v0.51.0 h1:94R/GTO7mt3/4wIKpcR5gkGmRLOuE/2hNGeWq/GBIFo= +golang.org/x/net v0.51.0/go.mod h1:aamm+2QF5ogm02fjy5Bb7CQ0WMt1/WVM7FtyaTLlA9Y= +golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.41.0 h1:Ivj+2Cp/ylzLiEU89QhWblYnOE9zerudt9Ftecq2C6k= +golang.org/x/sys v0.41.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks= +golang.org/x/text v0.34.0 h1:oL/Qq0Kdaqxa1KbNeMKwQq0reLCCaFtqu2eNuSeNHbk= +golang.org/x/text v0.34.0/go.mod h1:homfLqTYRFyVYemLBFl5GgL/DWEiH5wcsQ5gSh1yziA= +google.golang.org/protobuf v1.36.10 h1:AYd7cD/uASjIL6Q9LiTjz8JLcrh/88q5UObnmY3aOOE= +google.golang.org/protobuf v1.36.10/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk= +gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q= +gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/internal/hub.go b/internal/hub.go new file mode 100644 index 0000000..0a81c5e --- /dev/null +++ b/internal/hub.go @@ -0,0 +1,60 @@ +package internal + +import ( + "fmt" + "sync" + + "github.com/gorilla/websocket" +) + +// Client represents a connected user WebSocket connection. +type Client struct { + UID string + Conn *websocket.Conn +} + +// Hub manages connected clients and their WebSocket connections. +type Hub struct { + mu sync.RWMutex + clients map[string]*Client // uid -> client +} + +// NewHub creates a new Hub. +func NewHub() *Hub { + return &Hub{clients: make(map[string]*Client)} +} + +// Register adds a client connection. +func (h *Hub) Register(uid string, conn *websocket.Conn) { + h.mu.Lock() + defer h.mu.Unlock() + // Close existing connection if any + if old, ok := h.clients[uid]; ok { + old.Conn.Close() + } + h.clients[uid] = &Client{UID: uid, Conn: conn} +} + +// Unregister removes a client connection. +func (h *Hub) Unregister(uid string) { + h.mu.Lock() + defer h.mu.Unlock() + delete(h.clients, uid) +} + +// GetClient returns a client by uid. +func (h *Hub) GetClient(uid string) *Client { + h.mu.RLock() + defer h.mu.RUnlock() + return h.clients[uid] +} + +// PushToClient sends a JSON message to a specific client. +// Returns an error if the client is not connected. +func (h *Hub) PushToClient(uid string, msg []byte) error { + client := h.GetClient(uid) + if client == nil { + return fmt.Errorf("client %s not connected", uid) + } + return client.Conn.WriteMessage(websocket.TextMessage, msg) +} diff --git a/internal/wkapi/wkapi.go b/internal/wkapi/wkapi.go new file mode 100644 index 0000000..e63b996 --- /dev/null +++ b/internal/wkapi/wkapi.go @@ -0,0 +1,214 @@ +package wkapi + +import ( + "bytes" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "strings" + "time" +) + +// The WuKongIM access API can briefly reject a foreground write while the +// channel route converges to its current authority. A single client message +// number is idempotent, so retrying the same request is safe. +var sendRetryDelays = []time.Duration{ + 50 * time.Millisecond, + 100 * time.Millisecond, + 200 * time.Millisecond, + 400 * time.Millisecond, +} + +// Client wraps HTTP calls to WuKongIM 3.0 API. +type Client struct { + apiURL string + cli *http.Client +} + +// New creates a new WuKongIM API client. +func New(apiURL string) *Client { + return &Client{apiURL: apiURL, cli: &http.Client{}} +} + +// TokenResponse is the response from POST /user/token. +type TokenResponse struct { + UID string `json:"uid"` +} + +// RegisterToken registers a user token with WuKongIM. +func (c *Client) RegisterToken(uid, token string) error { + body := map[string]interface{}{ + "uid": uid, + "token": token, + "device_flag": 0, + "device_level": 1, + } + var resp TokenResponse + return c.post("/user/token", body, &resp) +} + +// UpsertChannel creates or updates a channel. +func (c *Client) UpsertChannel(channelID string, channelType uint8, subscribers []string) error { + body := map[string]interface{}{ + "channel_id": channelID, + "channel_type": channelType, + } + if len(subscribers) > 0 { + body["subscribers"] = subscribers + } + return c.post("/channel", body, nil) +} + +// SubscriberAdd adds subscribers to a channel. +func (c *Client) SubscriberAdd(channelID string, channelType uint8, uids []string) error { + body := map[string]interface{}{ + "channel_id": channelID, + "channel_type": channelType, + "reset": 0, + "subscribers": uids, + } + return c.post("/channel/subscriber_add", body, nil) +} + +// SendMessageRequest is the request body for POST /message/send. +type SendMessageRequest struct { + FromUID string `json:"from_uid"` + ChannelID string `json:"channel_id"` + ChannelType uint8 `json:"channel_type"` + ClientMsgNo string `json:"client_msg_no"` + Payload string `json:"payload"` // base64-encoded +} + +// SendMessageResponse is the response from POST /message/send. +type SendMessageResponse struct { + MessageID int64 `json:"message_id"` + MessageSeq uint64 `json:"message_seq"` + Reason uint8 `json:"reason"` +} + +// HTTPStatusError retains an API failure status so callers can distinguish a +// transient route response from a permanent validation or permission error. +type HTTPStatusError struct { + StatusCode int + Body string +} + +func (e *HTTPStatusError) Error() string { + return fmt.Sprintf("WK API returned status %d: %s", e.StatusCode, e.Body) +} + +// SendMessage sends a message via WuKongIM. +func (c *Client) SendMessage(req SendMessageRequest) (*SendMessageResponse, error) { + for attempt := 0; ; attempt++ { + var resp SendMessageResponse + err := c.post("/message/send", req, &resp) + if err == nil { + return &resp, nil + } + if !isRetryRequired(err) || attempt >= len(sendRetryDelays) { + return nil, err + } + time.Sleep(sendRetryDelays[attempt]) + } +} + +func isRetryRequired(err error) bool { + var statusErr *HTTPStatusError + return errors.As(err, &statusErr) && + statusErr.StatusCode == http.StatusServiceUnavailable && + strings.Contains(statusErr.Body, `"retry required"`) +} + +// SyncChannelMessagesRequest is the request body for POST /channel/messagesync. +type SyncChannelMessagesRequest struct { + LoginUID string `json:"login_uid"` + ChannelID string `json:"channel_id"` + ChannelType uint8 `json:"channel_type"` + StartMessageSeq uint64 `json:"start_message_seq"` + Limit int `json:"limit"` + PullMode int `json:"pull_mode"` +} + +// LegacyMessageResp is a message in the sync response (v3 legacy format). +type LegacyMessageResp struct { + Header map[string]interface{} `json:"header"` + Setting uint8 `json:"setting"` + MessageID uint64 `json:"message_id"` + ClientMsgNo string `json:"client_msg_no"` + MessageSeq uint64 `json:"message_seq"` + FromUID string `json:"from_uid"` + ChannelID string `json:"channel_id"` + ChannelType uint8 `json:"channel_type"` + Timestamp int32 `json:"timestamp"` + Payload string `json:"payload"` // base64-encoded +} + +// SyncChannelMessagesResponse is the response from POST /channel/messagesync. +type SyncChannelMessagesResponse struct { + StartMessageSeq uint64 `json:"start_message_seq"` + EndMessageSeq uint64 `json:"end_message_seq"` + More int `json:"more"` + Messages []LegacyMessageResp `json:"messages"` +} + +// SyncChannelMessages syncs messages from a channel. +func (c *Client) SyncChannelMessages(req SyncChannelMessagesRequest) (*SyncChannelMessagesResponse, error) { + var resp SyncChannelMessagesResponse + if err := c.post("/channel/messagesync", req, &resp); err != nil { + return nil, err + } + return &resp, nil +} + +// RouteResponse is the response from GET /route. +type RouteResponse struct { + TCPAddr string `json:"tcp_addr"` + WSAddr string `json:"ws_addr"` + WSSAddr string `json:"wss_addr"` +} + +// GetRoute gets the WebSocket route from WuKongIM. +func (c *Client) GetRoute() (*RouteResponse, error) { + var resp RouteResponse + if err := c.get("/route", &resp); err != nil { + return nil, err + } + return &resp, nil +} + +// --- internal helpers --- + +func (c *Client) post(path string, body interface{}, dest interface{}) error { + b, err := json.Marshal(body) + if err != nil { + return fmt.Errorf("marshal request: %w", err) + } + resp, err := c.cli.Post(c.apiURL+path, "application/json", bytes.NewReader(b)) + if err != nil { + return fmt.Errorf("http post %s: %w", path, err) + } + defer resp.Body.Close() + return c.decodeResponse(resp, dest) +} + +func (c *Client) get(path string, dest interface{}) error { + resp, err := c.cli.Get(c.apiURL + path) + if err != nil { + return fmt.Errorf("http get %s: %w", path, err) + } + defer resp.Body.Close() + return c.decodeResponse(resp, dest) +} + +func (c *Client) decodeResponse(resp *http.Response, dest interface{}) error { + if resp.StatusCode != http.StatusOK { + body, _ := io.ReadAll(resp.Body) + return &HTTPStatusError{StatusCode: resp.StatusCode, Body: string(body)} + } + if dest == nil { + return nil + } + return json.NewDecoder(resp.Body).Decode(dest) +} diff --git a/internal/wkapi/wkapi_test.go b/internal/wkapi/wkapi_test.go new file mode 100644 index 0000000..e0cc7a5 --- /dev/null +++ b/internal/wkapi/wkapi_test.go @@ -0,0 +1,69 @@ +package wkapi + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" +) + +func TestSendMessageRetriesRetryRequiredWithSameClientMessageNumber(t *testing.T) { + var attempts atomic.Int32 + var clientMsgNo string + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/message/send" { + t.Fatalf("path = %s, want /message/send", r.URL.Path) + } + var got SendMessageRequest + if err := json.NewDecoder(r.Body).Decode(&got); err != nil { + t.Fatalf("decode request: %v", err) + } + if clientMsgNo == "" { + clientMsgNo = got.ClientMsgNo + } else if got.ClientMsgNo != clientMsgNo { + t.Fatalf("client_msg_no = %q, want stable %q", got.ClientMsgNo, clientMsgNo) + } + if attempts.Add(1) <= 2 { + w.WriteHeader(http.StatusServiceUnavailable) + _, _ = w.Write([]byte(`{"error":"retry required"}`)) + return + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"message_id":101,"message_seq":7,"reason":1}`)) + })) + defer server.Close() + + client := New(server.URL) + response, err := client.SendMessage(SendMessageRequest{ + FromUID: "user_1", ChannelID: "agent_channel", ChannelType: 2, + ClientMsgNo: "lineup-idempotent-message", Payload: "aGVsbG8=", + }) + if err != nil { + t.Fatalf("SendMessage() error = %v", err) + } + if response.MessageID != 101 || response.MessageSeq != 7 { + t.Fatalf("SendMessage() response = %#v", response) + } + if got := attempts.Load(); got != 3 { + t.Fatalf("attempts = %d, want 3", got) + } +} + +func TestSendMessageDoesNotRetryPermanentAPIError(t *testing.T) { + var attempts atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + attempts.Add(1) + w.WriteHeader(http.StatusBadRequest) + _, _ = w.Write([]byte(`{"error":"invalid request"}`)) + })) + defer server.Close() + + _, err := New(server.URL).SendMessage(SendMessageRequest{ClientMsgNo: "invalid"}) + if err == nil { + t.Fatal("SendMessage() error = nil, want permanent error") + } + if got := attempts.Load(); got != 1 { + t.Fatalf("attempts = %d, want 1", got) + } +} diff --git a/main.go b/main.go new file mode 100644 index 0000000..9b48f48 --- /dev/null +++ b/main.go @@ -0,0 +1,90 @@ +package main + +import ( + "flag" + "fmt" + "net/http" + "os" + "strings" + + "github.com/gin-gonic/gin" + "github.com/spf13/viper" + + "github.com/lineup/app-server/modules" +) + +func main() { + var cfgFile string + flag.StringVar(&cfgFile, "config", "configs/lineup.yaml", "config file") + flag.Parse() + + vp := viper.New() + vp.SetConfigFile(cfgFile) + if err := vp.ReadInConfig(); err != nil { + fmt.Println("读取配置失败:", err) + os.Exit(1) + } + vp.SetEnvPrefix("LINEUP") + vp.SetEnvKeyReplacer(strings.NewReplacer(".", "_")) + vp.AutomaticEnv() + + gin.SetMode(gin.ReleaseMode) + if vp.GetString("mode") == "debug" { + gin.SetMode(gin.DebugMode) + } + + r := gin.Default() + r.Use(corsMiddleware(vp.GetStringSlice("client.allowedOrigins"))) + + // 静态文件服务 + r.Static("/chat", "./assets/chat") + // Markdown dependencies are served locally so the chat remains available on + // a private network and never executes content from a third-party CDN. + // They intentionally use their own prefix: /chat already owns a wildcard + // static route, which Gin cannot combine with nested exact file routes. + r.Static("/vendor/marked", "./node_modules/marked/lib") + r.Static("/vendor/dompurify", "./node_modules/dompurify/dist") + r.StaticFile("/", "./assets/chat/index.html") + + // 注册模块路由 + modules.SetupRoutes(r, vp) + + fmt.Printf("LineUp App Server 启动 → %s\n", vp.GetString("addr")) + fmt.Printf("WuKongIM API → %s\n", vp.GetString("wukongim.apiURL")) + r.Run(vp.GetString("addr")) +} + +// corsMiddleware permits only configured LineUp hosts to call the JSON API. +// The Web reference host is served same-origin and does not need an Origin +// header. Tauri uses its own WebView origin, so it must be explicitly allowed. +func corsMiddleware(origins []string) gin.HandlerFunc { + allowed := make(map[string]struct{}, len(origins)) + for _, origin := range origins { + if origin = strings.TrimSpace(origin); origin != "" { + allowed[origin] = struct{}{} + } + } + + return func(c *gin.Context) { + origin := c.GetHeader("Origin") + if origin != "" { + if _, ok := allowed[origin]; !ok { + if c.Request.Method == http.MethodOptions { + c.AbortWithStatus(http.StatusForbidden) + return + } + c.Next() + return + } + c.Header("Access-Control-Allow-Origin", origin) + c.Header("Access-Control-Allow-Methods", "GET, POST, OPTIONS") + c.Header("Access-Control-Allow-Headers", "Content-Type, Authorization, X-LineUp-Agent-Secret") + c.Header("Vary", "Origin") + } + if c.Request.Method == http.MethodOptions { + c.AbortWithStatus(http.StatusNoContent) + return + } + c.Next() + } +} diff --git a/modules/agent.go b/modules/agent.go new file mode 100644 index 0000000..c7fc6ba --- /dev/null +++ b/modules/agent.go @@ -0,0 +1,96 @@ +package modules + +import ( + "crypto/subtle" + "net/http" + "sync" + "time" + + "github.com/gin-gonic/gin" + "github.com/spf13/viper" +) + +// AgentModule is the app-server-facing control plane for external Agent +// adapters. Message delivery remains in MessageModule; this module makes the +// adapter's liveness and delivery cursor observable without exposing Hermes. +type AgentModule struct { + vp *viper.Viper + mu sync.RWMutex + entries map[string]agentHeartbeat +} + +type agentHeartbeat struct { + Status string `json:"status"` + LastMessageSeq uint64 `json:"last_message_seq"` + Detail string `json:"detail,omitempty"` + UpdatedAt time.Time `json:"updated_at"` +} + +type agentHeartbeatRequest struct { + Status string `json:"status" binding:"required"` + LastMessageSeq uint64 `json:"last_message_seq"` + Detail string `json:"detail"` +} + +func NewAgentModule(vp *viper.Viper) *AgentModule { + return &AgentModule{vp: vp, entries: make(map[string]agentHeartbeat)} +} + +func (m *AgentModule) RegisterRoutes(r *gin.Engine) { + r.GET("/agents/:uid/health", m.handleHealth) + r.POST("/agents/:uid/heartbeat", m.handleHeartbeat) +} + +func (m *AgentModule) configuredSecret() string { + return m.vp.GetString("agent.shared_secret") +} + +func (m *AgentModule) authorize(c *gin.Context) bool { + secret := m.configuredSecret() + if secret == "" { + c.JSON(http.StatusServiceUnavailable, gin.H{"error": "Agent heartbeat 未配置 shared_secret"}) + return false + } + provided := c.GetHeader("X-LineUp-Agent-Secret") + if subtle.ConstantTimeCompare([]byte(provided), []byte(secret)) != 1 { + c.JSON(http.StatusUnauthorized, gin.H{"error": "Agent heartbeat 认证失败"}) + return false + } + return true +} + +func (m *AgentModule) handleHeartbeat(c *gin.Context) { + if !m.authorize(c) { + return + } + var req agentHeartbeatRequest + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": "参数错误: status必填"}) + return + } + entry := agentHeartbeat{ + Status: req.Status, LastMessageSeq: req.LastMessageSeq, Detail: req.Detail, UpdatedAt: time.Now().UTC(), + } + m.mu.Lock() + m.entries[c.Param("uid")] = entry + m.mu.Unlock() + c.JSON(http.StatusOK, gin.H{"status": "ok", "updated_at": entry.UpdatedAt}) +} + +func (m *AgentModule) handleHealth(c *gin.Context) { + uid := c.Param("uid") + m.mu.RLock() + entry, found := m.entries[uid] + m.mu.RUnlock() + if !found { + c.JSON(http.StatusOK, gin.H{"uid": uid, "status": "offline", "known": false}) + return + } + // A worker normally reports once per poll. Keep a generous threshold so a + // slow Hermes turn does not look dead while it is still being supervised. + online := time.Since(entry.UpdatedAt) <= 45*time.Second + c.JSON(http.StatusOK, gin.H{ + "uid": uid, "status": entry.Status, "known": true, "online": online, + "last_message_seq": entry.LastMessageSeq, "detail": entry.Detail, "updated_at": entry.UpdatedAt, + }) +} diff --git a/modules/agent_test.go b/modules/agent_test.go new file mode 100644 index 0000000..db4efe0 --- /dev/null +++ b/modules/agent_test.go @@ -0,0 +1,47 @@ +package modules + +import ( + "bytes" + "net/http" + "net/http/httptest" + "testing" + + "github.com/gin-gonic/gin" + "github.com/spf13/viper" +) + +func TestAgentHeartbeatRequiresConfiguredSecret(t *testing.T) { + gin.SetMode(gin.TestMode) + vp := viper.New() + vp.Set("agent.shared_secret", "test-secret") + router := gin.New() + NewAgentModule(vp).RegisterRoutes(router) + + body := []byte(`{"status":"running","last_message_seq":42}`) + unauthorized := httptest.NewRequest(http.MethodPost, "/agents/agent_hermes_main/heartbeat", bytes.NewReader(body)) + unauthorized.Header.Set("Content-Type", "application/json") + unauthorizedResult := httptest.NewRecorder() + router.ServeHTTP(unauthorizedResult, unauthorized) + if unauthorizedResult.Code != http.StatusUnauthorized { + t.Fatalf("unauthorized status = %d, want %d", unauthorizedResult.Code, http.StatusUnauthorized) + } + + heartbeat := httptest.NewRequest(http.MethodPost, "/agents/agent_hermes_main/heartbeat", bytes.NewReader(body)) + heartbeat.Header.Set("Content-Type", "application/json") + heartbeat.Header.Set("X-LineUp-Agent-Secret", "test-secret") + heartbeatResult := httptest.NewRecorder() + router.ServeHTTP(heartbeatResult, heartbeat) + if heartbeatResult.Code != http.StatusOK { + t.Fatalf("heartbeat status = %d, body = %s", heartbeatResult.Code, heartbeatResult.Body.String()) + } + + health := httptest.NewRequest(http.MethodGet, "/agents/agent_hermes_main/health", nil) + healthResult := httptest.NewRecorder() + router.ServeHTTP(healthResult, health) + if healthResult.Code != http.StatusOK { + t.Fatalf("health status = %d, body = %s", healthResult.Code, healthResult.Body.String()) + } + if !bytes.Contains(healthResult.Body.Bytes(), []byte(`"online":true`)) { + t.Fatalf("health body = %s, want online=true", healthResult.Body.String()) + } +} diff --git a/modules/channel.go b/modules/channel.go new file mode 100644 index 0000000..a8ef6b7 --- /dev/null +++ b/modules/channel.go @@ -0,0 +1,75 @@ +package modules + +import ( + "net/http" + + "github.com/gin-gonic/gin" + "github.com/lineup/app-server/internal/wkapi" +) + +// ChannelModule handles channel creation and query. +type ChannelModule struct { + wk *wkapi.Client +} + +// NewChannelModule creates a new ChannelModule. +func NewChannelModule(wk *wkapi.Client) *ChannelModule { + return &ChannelModule{wk: wk} +} + +// RegisterRoutes registers channel routes. +func (m *ChannelModule) RegisterRoutes(r *gin.Engine) { + r.POST("/channels", m.handleCreate) + r.GET("/channels/:channel_id", m.handleGet) + r.POST("/channels/subscriber_add", m.handleSubscriberAdd) +} + +type createChannelRequest struct { + ChannelID string `json:"channel_id" binding:"required"` + ChannelType uint8 `json:"channel_type" binding:"required"` + Subscribers []string `json:"subscribers"` +} + +func (m *ChannelModule) handleCreate(c *gin.Context) { + var req createChannelRequest + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": "参数错误: channel_id和channel_type必填"}) + return + } + if err := m.wk.UpsertChannel(req.ChannelID, req.ChannelType, req.Subscribers); err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "创建频道失败: " + err.Error()}) + return + } + c.JSON(http.StatusOK, gin.H{ + "channel_id": req.ChannelID, + "channel_type": req.ChannelType, + }) +} + +func (m *ChannelModule) handleGet(c *gin.Context) { + channelID := c.Param("channel_id") + _ = channelID + c.JSON(http.StatusOK, gin.H{ + "channel_id": channelID, + "channel_type": 2, + }) +} + +type subscriberAddRequest struct { + ChannelID string `json:"channel_id" binding:"required"` + ChannelType uint8 `json:"channel_type" binding:"required"` + UIDs []string `json:"uids" binding:"required"` +} + +func (m *ChannelModule) handleSubscriberAdd(c *gin.Context) { + var req subscriberAddRequest + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": "参数错误: channel_id, channel_type, uids必填"}) + return + } + if err := m.wk.SubscriberAdd(req.ChannelID, req.ChannelType, req.UIDs); err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "添加订阅者失败: " + err.Error()}) + return + } + c.JSON(http.StatusOK, gin.H{"status": "ok"}) +} diff --git a/modules/client.go b/modules/client.go new file mode 100644 index 0000000..2650d68 --- /dev/null +++ b/modules/client.go @@ -0,0 +1,88 @@ +package modules + +import ( + "encoding/json" + "log" + "net/http" + "time" + + "github.com/gin-gonic/gin" + "github.com/gorilla/websocket" + "github.com/lineup/app-server/internal" + "github.com/lineup/app-server/internal/wkapi" +) + +var upgrader = websocket.Upgrader{ + CheckOrigin: func(r *http.Request) bool { return true }, +} + +// ClientModule manages user WebSocket connections for real-time push. +type ClientModule struct { + hub *internal.Hub + wk *wkapi.Client +} + +// NewClientModule creates a new ClientModule. +func NewClientModule(hub *internal.Hub, wk *wkapi.Client) *ClientModule { + return &ClientModule{hub: hub, wk: wk} +} + +// RegisterRoutes registers WebSocket routes. +func (m *ClientModule) RegisterRoutes(r *gin.Engine) { + r.GET("/ws", m.handleWS) +} + +func (m *ClientModule) handleWS(c *gin.Context) { + token := c.Query("token") + uid := c.Query("uid") + if token == "" || uid == "" { + c.JSON(http.StatusBadRequest, gin.H{"error": "参数错误: token和uid必填"}) + return + } + + // Verify token with WuKongIM (register if not exists, or validate) + if err := m.wk.RegisterToken(uid, token); err != nil { + c.JSON(http.StatusUnauthorized, gin.H{"error": "Token验证失败: " + err.Error()}) + return + } + + conn, err := upgrader.Upgrade(c.Writer, c.Request, nil) + if err != nil { + log.Printf("WebSocket升级失败: %v", err) + return + } + + m.hub.Register(uid, conn) + log.Printf("客户端连接: uid=%s", uid) + + // Send a welcome message + welcome, _ := json.Marshal(map[string]interface{}{ + "event": "connected", + "uid": uid, + }) + conn.WriteMessage(websocket.TextMessage, welcome) + + // Read loop (keep connection alive, handle pings) + go m.readLoop(uid, conn) +} + +func (m *ClientModule) readLoop(uid string, conn *websocket.Conn) { + defer func() { + m.hub.Unregister(uid) + conn.Close() + log.Printf("客户端断开: uid=%s", uid) + }() + + conn.SetReadDeadline(time.Now().Add(60 * time.Second)) + conn.SetPongHandler(func(string) error { + conn.SetReadDeadline(time.Now().Add(60 * time.Second)) + return nil + }) + + for { + _, _, err := conn.ReadMessage() + if err != nil { + break + } + } +} diff --git a/modules/message.go b/modules/message.go new file mode 100644 index 0000000..5e602b4 --- /dev/null +++ b/modules/message.go @@ -0,0 +1,143 @@ +package modules + +import ( + "encoding/base64" + "fmt" + "net/http" + "time" + + "github.com/gin-gonic/gin" + "github.com/lineup/app-server/internal/wkapi" +) + +// MessageModule handles message sending and syncing. +type MessageModule struct { + wk *wkapi.Client +} + +// NewMessageModule creates a new MessageModule. +func NewMessageModule(wk *wkapi.Client) *MessageModule { + return &MessageModule{wk: wk} +} + +// RegisterRoutes registers message routes. +func (m *MessageModule) RegisterRoutes(r *gin.Engine) { + r.POST("/messages/send", m.handleSend) + r.GET("/messages/sync", m.handleSync) + r.POST("/messages/sync", m.handleSyncPost) +} + +type sendMessageRequest struct { + FromUID string `json:"from_uid" binding:"required"` + ChannelID string `json:"channel_id" binding:"required"` + ChannelType uint8 `json:"channel_type" binding:"required"` + Payload string `json:"payload" binding:"required"` // raw text, will be base64-encoded +} + +func (m *MessageModule) handleSend(c *gin.Context) { + var req sendMessageRequest + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": "参数错误: from_uid, channel_id, channel_type, payload必填"}) + return + } + + // Base64-encode the payload for WuKongIM + encoded := base64.StdEncoding.EncodeToString([]byte(req.Payload)) + + resp, err := m.wk.SendMessage(wkapi.SendMessageRequest{ + FromUID: req.FromUID, + ChannelID: req.ChannelID, + ChannelType: req.ChannelType, + ClientMsgNo: fmt.Sprintf("msg_%d", time.Now().UnixNano()), + Payload: encoded, + }) + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "发送消息失败: " + err.Error()}) + return + } + c.JSON(http.StatusOK, gin.H{ + "message_id": resp.MessageID, + "message_seq": resp.MessageSeq, + }) +} + +type syncMessagesRequest struct { + ChannelID string `form:"channel_id" binding:"required"` + ChannelType uint8 `form:"channel_type" binding:"required"` + StartMessageSeq uint64 `form:"start_message_seq"` + Limit int `form:"limit"` +} + +func (m *MessageModule) handleSync(c *gin.Context) { + var req syncMessagesRequest + if err := c.ShouldBindQuery(&req); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": "参数错误: channel_id, channel_type必填"}) + return + } + m.doSync(c, req.ChannelID, req.ChannelType, req.StartMessageSeq, req.Limit, "") +} + +type syncMessagesPostRequest struct { + ChannelID string `json:"channel_id" binding:"required"` + ChannelType uint8 `json:"channel_type" binding:"required"` + LoginUID string `json:"login_uid"` + StartMessageSeq uint64 `json:"start_message_seq"` + Limit int `json:"limit"` +} + +func (m *MessageModule) handleSyncPost(c *gin.Context) { + var req syncMessagesPostRequest + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": "参数错误: channel_id, channel_type必填"}) + return + } + m.doSync(c, req.ChannelID, req.ChannelType, req.StartMessageSeq, req.Limit, req.LoginUID) +} + +func (m *MessageModule) doSync(c *gin.Context, channelID string, channelType uint8, startSeq uint64, limit int, loginUID string) { + if limit <= 0 { + limit = 20 + } + + resp, err := m.wk.SyncChannelMessages(wkapi.SyncChannelMessagesRequest{ + LoginUID: loginUID, + ChannelID: channelID, + ChannelType: channelType, + StartMessageSeq: startSeq, + Limit: limit, + PullMode: 1, + }) + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "同步消息失败: " + err.Error()}) + return + } + + // Decode base64 payloads for the response + type msgItem struct { + MessageID uint64 `json:"message_id"` + MessageSeq uint64 `json:"message_seq"` + FromUID string `json:"from_uid"` + ChannelID string `json:"channel_id"` + Payload string `json:"payload"` + Timestamp int32 `json:"timestamp"` + } + messages := make([]msgItem, 0, len(resp.Messages)) + for _, m := range resp.Messages { + decoded, _ := base64.StdEncoding.DecodeString(m.Payload) + messages = append(messages, msgItem{ + MessageID: m.MessageID, + MessageSeq: m.MessageSeq, + FromUID: m.FromUID, + ChannelID: m.ChannelID, + Payload: string(decoded), + Timestamp: m.Timestamp, + }) + } + + c.JSON(http.StatusOK, gin.H{ + "start_message_seq": resp.StartMessageSeq, + "end_message_seq": resp.EndMessageSeq, + "more": resp.More, + "messages": messages, + }) +} diff --git a/modules/modules.go b/modules/modules.go new file mode 100644 index 0000000..6368967 --- /dev/null +++ b/modules/modules.go @@ -0,0 +1,58 @@ +package modules + +import ( + "github.com/gin-gonic/gin" + "github.com/lineup/app-server/internal" + "github.com/lineup/app-server/internal/wkapi" + "github.com/spf13/viper" +) + +// SetupRoutes 注册所有模块路由 +func SetupRoutes(r *gin.Engine, vp *viper.Viper) { + // 共享依赖 + hub := internal.NewHub() + wk := wkapi.New(vp.GetString("wukongim.apiURL")) + + // 自动创建 Agent 频道 + agentChannelID := vp.GetString("agent.channelID") + agentUID := vp.GetString("agent.uid") + agentChannelType := uint8(vp.GetInt("agent.channelType")) + if agentChannelType == 0 { + agentChannelType = 2 + } + if agentChannelID != "" && agentUID != "" { + if err := wk.UpsertChannel(agentChannelID, agentChannelType, []string{agentUID}); err != nil { + // ignore upsert error, channel may already exist + _ = err + } + } + + // 健康检查 + r.GET("/healthz", func(c *gin.Context) { + c.JSON(200, gin.H{"status": "ok"}) + }) + + // 用户模块 + userModule := &UserModule{vp: vp, wk: wk} + userModule.RegisterRoutes(r) + + // Channel 模块 + channelModule := NewChannelModule(wk) + channelModule.RegisterRoutes(r) + + // Message 模块 + messageModule := NewMessageModule(wk) + messageModule.RegisterRoutes(r) + + // Agent 控制面：外部 Hermes Adapter 的存活与消费游标。 + agentModule := NewAgentModule(vp) + agentModule.RegisterRoutes(r) + + // Webhook 模块 + webhookModule := NewWebhookModule(hub) + webhookModule.RegisterRoutes(r) + + // Client WebSocket 模块 + clientModule := NewClientModule(hub, wk) + clientModule.RegisterRoutes(r) +} diff --git a/modules/user.go b/modules/user.go new file mode 100644 index 0000000..d775d95 --- /dev/null +++ b/modules/user.go @@ -0,0 +1,95 @@ +package modules + +import ( + "crypto/rand" + "encoding/hex" + "fmt" + "net/http" + + "github.com/gin-gonic/gin" + "github.com/lineup/app-server/internal/wkapi" + "github.com/spf13/viper" +) + +// UserModule 用户模块: 登录 / 注册 +type UserModule struct { + vp *viper.Viper + wk *wkapi.Client +} + +// RegisterRoutes 注册路由 +func (m *UserModule) RegisterRoutes(r *gin.Engine) { + r.POST("/login", m.handleLogin) +} + +// LoginRequest 登录请求 +type LoginRequest struct { + Phone string `json:"phone" binding:"required"` + Code string `json:"code" binding:"required"` +} + +// LoginResponse 登录响应 +type LoginResponse struct { + UID string `json:"uid"` + Token string `json:"token"` + IMAddr string `json:"im_addr"` // WuKongIM WS地址 + AgentUID string `json:"agent_uid"` // Agent的UID + ChannelID string `json:"channel_id"` // Agent频道ID + ChannelType uint8 `json:"channel_type"` // Agent频道类型 +} + +// handleLogin 处理登录请求 +func (m *UserModule) handleLogin(c *gin.Context) { + var req LoginRequest + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": "参数错误: phone和code必填"}) + return + } + + // 生成用户 UID + uid := fmt.Sprintf("user_%s", req.Phone) + + // 生成 Token (随机32字节) + token := generateToken() + + // 注册 Token 到 WuKongIM + if err := m.wk.RegisterToken(uid, token); err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": fmt.Sprintf("注册Token失败: %v", err)}) + return + } + + // 确保 Agent 频道存在并将用户加入订阅者 + agentChannelID := m.vp.GetString("agent.channelID") + agentUID := m.vp.GetString("agent.uid") + agentChannelType := uint8(m.vp.GetInt("agent.channelType")) + if agentChannelType == 0 { + agentChannelType = 2 + } + // 创建/更新频道，确保 Agent 和用户都是订阅者 + m.wk.UpsertChannel(agentChannelID, agentChannelType, []string{agentUID}) + // 将用户加入频道订阅者 + m.wk.SubscriberAdd(agentChannelID, agentChannelType, []string{uid}) + + // 获取 WebSocket 地址 + route, err := m.wk.GetRoute() + wsAddr := m.vp.GetString("wukongim.wsAddr") + if err == nil && route.WSAddr != "" { + wsAddr = route.WSAddr + } + + c.JSON(http.StatusOK, LoginResponse{ + UID: uid, + Token: token, + IMAddr: wsAddr, + AgentUID: m.vp.GetString("agent.uid"), + ChannelID: m.vp.GetString("agent.channelID"), + ChannelType: uint8(m.vp.GetInt("agent.channelType")), + }) +} + +// generateToken 生成32字节随机Token +func generateToken() string { + b := make([]byte, 32) + rand.Read(b) + return hex.EncodeToString(b) +} diff --git a/modules/webhook.go b/modules/webhook.go new file mode 100644 index 0000000..470d088 --- /dev/null +++ b/modules/webhook.go @@ -0,0 +1,151 @@ +package modules + +import ( + "encoding/base64" + "encoding/json" + "fmt" + "io" + "net/http" + + "github.com/gin-gonic/gin" + "github.com/lineup/app-server/internal" +) + +// WebhookModule receives WuKongIM 3.0 HTTP webhook callbacks. +type WebhookModule struct { + hub *internal.Hub +} + +// NewWebhookModule creates a new WebhookModule. +func NewWebhookModule(hub *internal.Hub) *WebhookModule { + return &WebhookModule{hub: hub} +} + +// RegisterRoutes registers webhook routes. +func (m *WebhookModule) RegisterRoutes(r *gin.Engine) { + r.POST("/v1/webhook", m.handleWebhook) +} + +// wkMsgResp mirrors WuKongIM 3.0's msg.notify message format. +type wkMsgResp struct { + Header wkMsgHeader `json:"header"` + Setting uint8 `json:"setting"` + MessageID uint64 `json:"message_id"` + MessageSeq uint64 `json:"message_seq"` + FromUID string `json:"from_uid"` + ChannelID string `json:"channel_id"` + ChannelType uint8 `json:"channel_type"` + Timestamp int32 `json:"timestamp"` + Payload string `json:"payload"` // base64-encoded +} + +type wkMsgHeader struct { + NoPersist int `json:"no_persist"` + RedDot int `json:"red_dot"` + SyncOnce int `json:"sync_once"` +} + +// wkOfflineResp mirrors WuKongIM 3.0's msg.offline format. +type wkOfflineResp struct { + wkMsgResp + ToUIDs []string `json:"to_uids"` + Compress string `json:"compress"` + CompressToUIDs string `json:"compress_to_uids"` + SourceID int64 `json:"source_id"` +} + +func (m *WebhookModule) handleWebhook(c *gin.Context) { + event := c.Query("event") + body, err := io.ReadAll(c.Request.Body) + if err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": "读取请求体失败"}) + return + } + + switch event { + case "msg.notify": + m.handleMsgNotify(body) + case "msg.offline": + m.handleMsgOffline(body) + case "user.onlinestatus": + m.handleOnlineStatus(body) + default: + // Unknown event, still return 200 per WK webhook contract + } + + c.JSON(http.StatusOK, gin.H{"status": "ok"}) +} + +func (m *WebhookModule) handleMsgNotify(data []byte) { + var msgs []wkMsgResp + if err := json.Unmarshal(data, &msgs); err != nil { + return + } + for _, msg := range msgs { + m.pushToClient(msg.ChannelID, msg) + } +} + +func (m *WebhookModule) handleMsgOffline(data []byte) { + var msg wkOfflineResp + if err := json.Unmarshal(data, &msg); err != nil { + return + } + // Push to each offline recipient + for _, uid := range msg.ToUIDs { + m.pushToClient(uid, msg.wkMsgResp) + } + // Handle gzip-compressed UIDs + if msg.Compress == "gzip" && len(msg.CompressToUIDs) > 0 { + decoded, err := base64.StdEncoding.DecodeString(msg.CompressToUIDs) + if err == nil { + var uids []string + if json.Unmarshal(decoded, &uids) == nil { + for _, uid := range uids { + m.pushToClient(uid, msg.wkMsgResp) + } + } + } + } +} + +func (m *WebhookModule) handleOnlineStatus(data []byte) { + // Online status format: ["uid-flag-status", ...] + var statuses []string + if err := json.Unmarshal(data, &statuses); err != nil { + return + } + // For now, just log. Could be used to notify clients. + _ = statuses +} + +// pushToClient decodes the payload and pushes to the connected client. +func (m *WebhookModule) pushToClient(uid string, msg wkMsgResp) { + if m.hub == nil { + return + } + decoded, _ := base64.StdEncoding.DecodeString(msg.Payload) + pushMsg := map[string]interface{}{ + "event": "message", + "message_id": msg.MessageID, + "message_seq": msg.MessageSeq, + "from_uid": msg.FromUID, + "channel_id": msg.ChannelID, + "channel_type": msg.ChannelType, + "payload": string(decoded), + "timestamp": msg.Timestamp, + } + pushData, _ := json.Marshal(pushMsg) + if err := m.hub.PushToClient(uid, pushData); err != nil { + // Client not connected, that's OK + _ = err + } + // Also push to the other participant if it's a person channel + // This ensures both sides get the message + if msg.ChannelType == 1 && msg.FromUID != uid { + _ = m.hub.PushToClient(msg.FromUID, pushData) + } +} + +// Ensure fmt is used (for potential future logging) +var _ = fmt.Sprintf diff --git a/package-lock.json b/package-lock.json new file mode 100644 index 0000000..be8194b --- /dev/null +++ b/package-lock.json @@ -0,0 +1,67 @@ +{ + "name": "lineup-app-server", + "version": "1.0.0", + "lockfileVersion": 3, + "requires": true, + "packages": { + "": { + "name": "lineup-app-server", + "version": "1.0.0", + "license": "ISC", + "dependencies": { + "dompurify": "^3.4.12", + "marked": "^15.0.12", + "ws": "^8.21.1" + } + }, + "node_modules/@types/trusted-types": { + "version": "2.0.7", + "resolved": "https://registry.npmjs.org/@types/trusted-types/-/trusted-types-2.0.7.tgz", + "integrity": "sha512-ScaPdn1dQczgbl0QFTeTOmVHFULt394XJgOQNoyVhZ6r2vLnMLJfBPd53SB52T/3G36VI1/g2MZaX0cwDuXsfw==", + "license": "MIT", + "optional": true + }, + "node_modules/dompurify": { + "version": "3.4.12", + "resolved": "https://registry.npmjs.org/dompurify/-/dompurify-3.4.12.tgz", + "integrity": "sha512-zQvGet8Z2sWbQhCmfFz/T5QWH2oBmjnqK3qvOjaqaNLrLEF912WamU+ohnTp0TCep/MFVHpdJuCZEdFOdTnEFg==", + "license": "(MPL-2.0 OR Apache-2.0)", + "optionalDependencies": { + "@types/trusted-types": "^2.0.7" + } + }, + "node_modules/marked": { + "version": "15.0.12", + "resolved": "https://registry.npmjs.org/marked/-/marked-15.0.12.tgz", + "integrity": "sha512-8dD6FusOQSrpv9Z1rdNMdlSgQOIP880DHqnohobOmYLElGEqAL/JvxvuxZO16r4HtjTlfPRDC1hbvxC9dPN2nA==", + "license": "MIT", + "bin": { + "marked": "bin/marked.js" + }, + "engines": { + "node": ">= 18" + } + }, + "node_modules/ws": { + "version": "8.21.1", + "resolved": "https://registry.npmjs.org/ws/-/ws-8.21.1.tgz", + "integrity": "sha512-+0NTnW77fFN/DjQi6k/Sq/Yvk4Sgajw7urW8V+asjXnRgDs9gyGkdb7EzgfhA4goXsRIZKE28fzIXBHEzhuiWw==", + "license": "MIT", + "engines": { + "node": ">=10.0.0" + }, + "peerDependencies": { + "bufferutil": "^4.0.1", + "utf-8-validate": ">=5.0.2" + }, + "peerDependenciesMeta": { + "bufferutil": { + "optional": true + }, + "utf-8-validate": { + "optional": true + } + } + } + } +} diff --git a/package.json b/package.json new file mode 100644 index 0000000..591751c --- /dev/null +++ b/package.json @@ -0,0 +1,18 @@ +{ + "name": "lineup-app-server", + "version": "1.0.0", + "description": "", + "main": "index.js", + "scripts": { + "test": "node --test assets/chat/*.test.js" + }, + "keywords": [], + "author": "", + "license": "ISC", + "type": "commonjs", + "dependencies": { + "dompurify": "^3.4.12", + "marked": "^15.0.12", + "ws": "^8.21.1" + } +} diff --git a/run-server.sh b/run-server.sh new file mode 100755 index 0000000..6aa11d5 --- /dev/null +++ b/run-server.sh @@ -0,0 +1,51 @@ +#!/usr/bin/env bash +# Minimal local supervisor for the LineUp app server. It is intentionally the +# only process manager for this binary; the Hermes Adapter has its own script. +set -euo pipefail + +DIR="$(cd "$(dirname "$0")" && pwd)" +BIN="$DIR/bin/lineup-server" +CONFIG="$DIR/configs/lineup.yaml" +LOG="$DIR/appserver.log" +PIDFILE="$DIR/lineup-server.pid" +RUNTIME_ENV="$DIR/.env" + +# Deployment secrets live beside the service, never in lineup.yaml. Viper maps +# LINEUP_AGENT_SHARED_SECRET to agent.shared_secret automatically. +if [[ -f "$RUNTIME_ENV" ]]; then + set -a + # shellcheck disable=SC1090 + source "$RUNTIME_ENV" + set +a +fi + +is_running() { + [[ -f "$PIDFILE" ]] && kill -0 "$(<"$PIDFILE")" 2>/dev/null +} + +start() { + if is_running; then + echo "LineUp App Server 已在运行 PID=$(<"$PIDFILE")" + return + fi + cd "$DIR" + nohup setsid "$BIN" -config "$CONFIG" >> "$LOG" 2>&1 < /dev/null & + echo "$!" > "$PIDFILE" + echo "LineUp App Server 已启动 PID=$(<"$PIDFILE")" +} + +stop() { + if is_running; then + kill "$(<"$PIDFILE")" + echo "LineUp App Server 已停止" + fi + rm -f "$PIDFILE" +} + +case "${1:-start}" in + start) start ;; + stop) stop ;; + restart) stop; start ;; + status) is_running && echo "LineUp App Server 运行中 PID=$(<"$PIDFILE")" || echo "LineUp App Server 未运行" ;; + *) echo "用法: $0 {start|stop|restart|status}"; exit 2 ;; +esac