// 'use strict';
// require('dotenv').config();
// const crypto = require('node:crypto');
// const express = require('express');
// const WebSocket = require('ws');
// const axios = require('axios');
// const {
// S3Client,
// PutObjectCommand,
// GetObjectCommand,
// ListObjectsV2Command,
// DeleteObjectCommand
// } = require('@aws-sdk/client-s3');
// // ============================================================
// // 0. 核心配置
// // ============================================================
// const PORT = process.env.PORT || 7860;
// const {
// OPENWEBUI_BASE_URL,
// OPENWEBUI_API_KEY,
// OPENCLAW_WS_URL,
// OPENCLAW_TOKEN,
// BOT_USERNAME = 'openman',
// POLL_INTERVAL = 5000,
// // S3 配置
// STORAGE_EXTERNAL_ENDPOINT,
// STORAGE_PUBLIC_BUCKET,
// STORAGE_REGION,
// STORAGE_ACCESS_KEY_ID,
// STORAGE_SECRET_ACCESS_KEY,
// STORAGE_S3_FORCE_PATH_STYLE
// } = process.env;
// // S3 客户端
// const s3 = new S3Client({
// region: STORAGE_REGION || 'us-east-1',
// endpoint: STORAGE_EXTERNAL_ENDPOINT,
// forcePathStyle: STORAGE_S3_FORCE_PATH_STYLE === 'true',
// credentials: {
// accessKeyId: STORAGE_ACCESS_KEY_ID,
// secretAccessKey: STORAGE_SECRET_ACCESS_KEY
// }
// });
// const S3_PREFIX_CONFIG = 'openclaw/channels.json';
// const S3_PREFIX_UPLOADS = 'openclaw/uploads/';
// // 日志系统
// const logBuffer = [];
// function addLog(type, msg) {
// const time = new Date().toLocaleTimeString('zh-CN', { hour12: false });
// const line = `[${time}] ${type === 'err' ? '❌ ' : ''}${msg}`;
// console[type === 'err' ? 'error' : 'log'](line);
// logBuffer.unshift(line);
// if (logBuffer.length > 100) logBuffer.pop();
// }
// function log(msg) { addLog('info', msg); }
// function err(msg) { addLog('err', msg); }
// const channelRuntime = new Map();
// let ws = null;
// let isClawReady = false;
// let isReconnecting = false;
// let pingInterval = null;
// let reconnectTimer = null;
// const pendingRequests = new Map();
// const webuiApi = axios.create({
// baseURL: OPENWEBUI_BASE_URL,
// headers: { 'Content-Type': 'application/json', 'Authorization': `Bearer ${OPENWEBUI_API_KEY}` },
// timeout: 10000
// });
// const fileDownloader = axios.create({
// baseURL: OPENWEBUI_BASE_URL,
// headers: { 'Authorization': `Bearer ${OPENWEBUI_API_KEY}` },
// responseType: 'arraybuffer',
// timeout: 30000
// });
// // ============================================================
// // 1. S3 核心功能
// // ============================================================
// async function streamToString(stream) {
// return new Promise((resolve, reject) => {
// const chunks = [];
// stream.on('data', (chunk) => chunks.push(chunk));
// stream.on('error', reject);
// stream.on('end', () => resolve(Buffer.concat(chunks).toString('utf8')));
// });
// }
// async function loadChannelsFromS3() {
// try {
// log('☁️ [S3] 读取配置...');
// const command = new GetObjectCommand({ Bucket: STORAGE_PUBLIC_BUCKET, Key: S3_PREFIX_CONFIG });
// const response = await s3.send(command);
// const str = await streamToString(response.Body);
// const data = JSON.parse(str);
// if (data && Array.isArray(data.channels)) {
// data.channels.forEach(ch => {
// if (!channelRuntime.has(ch.id)) {
// channelRuntime.set(ch.id, { id: ch.id, name: ch.name, lastMsgId: null, syncing: false });
// initChannelBaseline(ch.id);
// }
// });
// log(`✅ [S3] 载入 ${data.channels.length} 个频道`);
// }
// } catch (e) {
// if (e.name === 'NoSuchKey') log('ℹ️ [S3] 无配置文件，使用默认');
// else err(`S3读取错误: ${e.message}`);
// }
// }
// async function saveChannelsToS3() {
// try {
// const list = Array.from(channelRuntime.values()).map(s => ({ id: s.id, name: s.name }));
// const json = JSON.stringify({ version: "2.0", savedAt: new Date().toISOString(), channels: list }, null, 2);
// await s3.send(new PutObjectCommand({
// Bucket: STORAGE_PUBLIC_BUCKET,
// Key: S3_PREFIX_CONFIG,
// Body: json,
// ContentType: 'application/json'
// }));
// } catch (e) { err(`S3保存错误: ${e.message}`); }
// }
// async function uploadToS3(filename, buffer, mimeType) {
// const key = S3_PREFIX_UPLOADS + filename;
// try {
// log(`☁️ [Debug] 开始上传S3: ${filename} (${buffer.length} bytes)`);
// await s3.send(new PutObjectCommand({
// Bucket: STORAGE_PUBLIC_BUCKET,
// Key: key,
// Body: buffer,
// ContentType: mimeType || 'application/octet-stream',
// ACL: 'public-read'
// }));
// let baseUrl = STORAGE_EXTERNAL_ENDPOINT;
// if (!baseUrl.endsWith('/')) baseUrl += '/';
// const url = `${baseUrl}${STORAGE_PUBLIC_BUCKET}/${key}`;
// log(`✅ [Debug] S3上传成功: ${url}`);
// return url;
// } catch (e) {
// err(`❌ S3上传失败 [${filename}]: ${e.message}`);
// return null;
// }
// }
// async function cleanOldS3Files() {
// log('🧹 [S3] 检查过期文件...');
// try {
// const threeDaysAgo = Date.now() - (3 * 24 * 60 * 60 * 1000);
// let continuationToken = null;
// let deletedCount = 0;
// do {
// const result = await s3.send(new ListObjectsV2Command({
// Bucket: STORAGE_PUBLIC_BUCKET,
// Prefix: S3_PREFIX_UPLOADS,
// ContinuationToken: continuationToken
// }));
// if (result.Contents) {
// for (const item of result.Contents) {
// if (item.LastModified && item.LastModified.getTime() < threeDaysAgo) {
// await s3.send(new DeleteObjectCommand({ Bucket: STORAGE_PUBLIC_BUCKET, Key: item.Key }));
// deletedCount++;
// }
// }
// }
// continuationToken = result.NextContinuationToken;
// } while (continuationToken);
// if(deletedCount > 0) log(`🗑️ 已清理 ${deletedCount} 个旧文件`);
// } catch (e) { err(`清理失败: ${e.message}`); }
// }
// // ============================================================
// // 2. Web 界面
// // ============================================================
// const app = express();
// app.use(express.json());
// const HTML_PAGE = `
//
//
//
// Bridge Console (S3)
//
//
//
//
//
Bridge Console S3 + NewAPI
//
Claw: ...
//
Bridge Console No History Mode
Claw: ...
`;
app.get('/', (req, res) => res.send(HTML_PAGE));
app.get('/api/channels', (req, res) => res.json(Array.from(channelRuntime.values())));
app.get('/api/logs', (req, res) => res.json(logBuffer));
app.post('/api/channels', async (req, res) => {
const { id, name } = req.body;
if (!channelRuntime.has(id)) {
channelRuntime.set(id, { id, name: name||id, lastMsgId: null, syncing: false });
await saveChannelsToS3();
initChannelBaseline(id);
}
res.json({success:true});
});
app.delete('/api/channels/:id', async (req, res) => {
if (channelRuntime.delete(req.params.id)) await saveChannelsToS3();
res.json({success:true});
});
app.get('/health', (req, res) => res.json({ status:'ok', wsConnected:isClawReady }));
app.listen(PORT, async () => {
log(`🚀 服务启动: ${PORT}`);
await init();
});
// ============================================================
// 3. 业务逻辑
// ============================================================
async function init() {
try {
await loadChannelsFromS3();
connectOpenClaw();
setInterval(cleanOldS3Files, 3600 * 1000);
cleanOldS3Files();
} catch (e) { err(e.message); }
}
async function initChannelBaseline(id) {
const state = channelRuntime.get(id);
if (!state) return;
try {
// 🛑 核心保护：启动时直接获取最新一条消息 ID。
// 不管以前发生过什么，我们只认这一刻最新的消息作为起点。
const res = await webuiApi.get(`/api/v1/channels/${id}/messages?limit=1`);
if (res.data && res.data.length > 0) {
state.lastMsgId = res.data[0].id; // 锁定最新ID
log(`✅ [${state.name}] 初始化完成，锚点锁定: ${state.lastMsgId} (不发送历史)`);
} else {
state.lastMsgId = "EMPTY_START";
log(`✅ [${state.name}] 频道为空，等待第一条消息...`);
}
} catch { setTimeout(() => initChannelBaseline(id), 5000); }
}
// WS 连接逻辑
const { publicKey, privateKey } = crypto.generateKeyPairSync('ed25519');
const rawPublicKey = publicKey.export({ type: 'spki', format: 'der' }).subarray(12);
const deviceId = crypto.createHash('sha256').update(rawPublicKey).digest('hex');
const toB64Url = (buf) => buf.toString('base64').replace(/\+/g, '-').replace(/\//g, '_').replace(/=+$/, '');
function cleanupConnection() {
if (pingInterval) clearInterval(pingInterval);
if (ws) { ws.removeAllListeners(); try { ws.terminate(); } catch {} ws = null; }
isClawReady = false;
for (const state of channelRuntime.values()) state.syncing = false;
}
function connectOpenClaw(retryCount = 0) {
if (isReconnecting || ws) return;
isReconnecting = true;
log(`🔌 连接 OpenClaw (${retryCount + 1})...`);
try {
ws = new WebSocket(OPENCLAW_WS_URL.replace(/\/$/, '') + '/ws');
ws.on('open', () => {
log('✅ WS 握手成功');
isReconnecting = false;
ws.isAlive = true;
ws.on('pong', () => { ws.isAlive = true; });
pingInterval = setInterval(() => {
if (ws.isAlive === false) { err('心跳超时'); return ws.terminate(); }
ws.isAlive = false;
ws.ping();
}, 10000);
});
ws.on('message', handleClawMessage);
ws.on('close', (code) => { log(`🔌 断开 (${code})`); handleDisconnect(retryCount); });
ws.on('error', (e) => err(`WS错误: ${e.message}`));
} catch (e) { err(e.message); handleDisconnect(retryCount); }
}
function handleDisconnect(retryCount) {
cleanupConnection();
isReconnecting = false;
const delay = Math.min(5000 * Math.pow(1.5, retryCount), 60000);
if (reconnectTimer) clearTimeout(reconnectTimer);
reconnectTimer = setTimeout(() => { connectOpenClaw(retryCount + 1); }, delay);
}
function handleClawMessage(data) {
let msg;
try { msg = JSON.parse(data.toString()); } catch { return; }
if (msg.event === 'connect.challenge') {
const { nonce } = msg.payload;
const signedAt = Date.now();
const payload = ['v2', deviceId, 'openclaw-control-ui', 'webchat', 'operator', "operator.admin,operator.approvals,operator.pairing", String(signedAt), OPENCLAW_TOKEN||'', nonce].join('|');
const signature = crypto.sign(null, Buffer.from(payload), privateKey);
ws.send(JSON.stringify({
type: 'req', id: crypto.randomUUID(), method: 'connect',
params: {
minProtocol: 3, maxProtocol: 3,
client: { id: 'openclaw-control-ui', version: 'dev', platform: 'Win32', mode: 'webchat' },
device: { id: deviceId, publicKey: toB64Url(rawPublicKey), signature: toB64Url(signature), signedAt, nonce },
auth: { token: OPENCLAW_TOKEN },
role: 'operator', scopes: ["operator.admin","operator.approvals","operator.pairing"]
}
}));
}
if (msg.payload?.type === 'hello-ok') { log('🎉 OpenClaw 认证成功'); isClawReady = true; }
if (msg.type === 'res' && msg.id && pendingRequests.has(msg.id)) {
const handler = pendingRequests.get(msg.id);
if (Array.isArray(handler)) handler[0](msg.payload);
pendingRequests.delete(msg.id);
}
}
function fetchClawHistory(cid, uid) {
return new Promise((resolve) => {
if (!isClawReady) return resolve([]);
const id = crypto.randomUUID();
const t = setTimeout(() => { pendingRequests.delete(id); resolve([]); }, 5000);
pendingRequests.set(id, [(p) => { clearTimeout(t); resolve(p.messages || []); }, () => resolve([])]);
try { ws.send(JSON.stringify({ type:'req', id, method:'chat.history', params:{ sessionKey:`agent:main:${cid}`, limit:5 } })); }
catch { resolve([]); }
});
}
function triggerAgent(cid, uid, name, text, isCommand) {
if (!isClawReady) return;
const finalMsg = isCommand ? text : `${name} 说: ${text}`;
ws.send(JSON.stringify({
type: 'req', id: crypto.randomUUID(), method: isCommand ? 'chat.send' : 'agent',
params: { sessionKey: `agent:main:${cid}`, message: finalMsg, idempotencyKey: crypto.randomUUID(), deliver: isCommand?false:undefined }
}));
log(`🚀 [发送] ${name}: ${text.slice(0,20)}...`);
}
// ============================================================
// 4. 同步逻辑 (严防历史消息泄漏版)
// ============================================================
async function syncChannel(cid) {
const state = channelRuntime.get(cid);
if (!state || !state.lastMsgId || !isClawReady) return;
if (state.syncing) {
if (Date.now() - (state.syncStartTime || 0) > 60000) { log(`⚠️ [${state.name}] 超时复位`); state.syncing = false; }
else return;
}
state.syncing = true;
state.syncStartTime = Date.now();
try {
const { data: msgs } = await webuiApi.get(`/api/v1/channels/${cid}/messages?skip=0&limit=10`);
if (!msgs?.length) throw new Error("No messages");
// --- 1. 回信检测 (保持不变) ---
const webuiFingerprints = new Set(msgs.map(m => (m.content || m.data?.content || '').trim()).filter(Boolean));
const activeUsers = new Map();
msgs.slice(0, 5).forEach(m => { if(m.user_id) activeUsers.set(m.user_id, {id: m.user_id, name: m.user?.name||'User'}); });
for (const u of activeUsers.values()) {
await new Promise(r => setTimeout(r, 200));
const history = await fetchClawHistory(cid, u.id);
const replies = history.filter(h => h.role === 'assistant').slice(-2);
for (const r of replies) {
const txt = r.content?.[0]?.text?.trim();
if (txt && !webuiFingerprints.has(txt)) {
log(`📥 [回复] -> ${u.name}`);
await webuiApi.post(`/api/v1/channels/${cid}/messages/post`, { content: txt, data:{files:[]}, reply_to_id:null });
webuiFingerprints.add(txt);
}
}
}
// --- 2. 新消息处理 (WebUI -> Claw) ---
const sorted = msgs.reverse();
// 🔴 关键修复逻辑：计算新消息
let newMsgs = [];
const startIdx = sorted.findIndex(m => m.id === state.lastMsgId);
if (state.lastMsgId === "EMPTY_START") {
// 频道刚开始是空的，现在有消息了，全部视为新消息
newMsgs = sorted;
} else if (startIdx !== -1) {
// 找到了上次的锚点，正常截取之后的消息
newMsgs = sorted.slice(startIdx + 1);
} else {
// ⚠️ 严重情况：找不到锚点 (说明被删除了)。
// 绝对不能发送 sorted，否则就是炸弹。
// 策略：重置锚点为当前最新，静默跳过。
if (sorted.length > 0) {
const newestId = sorted[sorted.length - 1].id;
log(`⚠️ [${state.name}] 锚点丢失(已删)，重置锚点 -> ${newestId}，跳过历史同步。`);
state.lastMsgId = newestId;
}
newMsgs = []; // 强制为空
}
if (newMsgs.length > 0) {
state.lastMsgId = sorted[sorted.length - 1].id; // 更新锚点
for (const m of newMsgs) {
const sender = m.user?.name || '';
if (sender.toLowerCase() === BOT_USERNAME.toLowerCase()) continue;
let txt = (m.content || m.data?.content || '').replace(/<@U:[^>]+>/g,'').replace(new RegExp(`@${BOT_USERNAME}`,'gi'),'').trim();
let atts = [];
// 附件处理 (懒加载兼容逻辑)
let rawFiles = [];
if (m.data === true) {
log(`📡 [Debug] 发现懒加载数据，请求详情: ${m.id}`);
try {
const { data: detailData } = await webuiApi.get(`/api/v1/channels/${cid}/messages/${m.id}/data`);
if (detailData && detailData.files) rawFiles = detailData.files;
} catch (e) { err(` ❌ 详情获取失败: ${e.message}`); }
} else if (typeof m.data === 'object' && m.data?.files) {
rawFiles = m.data.files;
}
if (rawFiles.length > 0) {
log(`🔎 [Debug] 处理 ${rawFiles.length} 个附件...`);
for (const fItem of rawFiles) {
try {
let buffer, filename, mimeType;
const f = fItem.file || fItem;
// Base64
if (fItem.url && fItem.url.startsWith('data:')) {
const matches = fItem.url.match(/^data:([a-zA-Z0-9/+-]+);base64,(.+)$/);
if (matches) {
mimeType = matches[1];
const extMap = {'image/jpeg':'jpg', 'image/png':'png', 'image/gif':'gif', 'image/webp':'webp'};
const ext = extMap[mimeType] || 'bin';
buffer = Buffer.from(matches[2], 'base64');
filename = `img_${Date.now()}_${Math.random().toString(36).slice(2)}.${ext}`;
}
}
// ID 下载
else if (f.id) {
log(`⬇️ [Debug] 下载文件: ${f.filename}`);
const { data } = await fileDownloader.get(`/api/v1/files/${f.id}/content`);
buffer = data;
filename = `${Date.now()}_${(f.filename || 'file').replace(/[^a-zA-Z0-9.-]/g, '_')}`;
mimeType = f.content_type || fItem.content_type || 'application/octet-stream';
}
if (buffer && filename) {
const s3Url = await uploadToS3(filename, buffer, mimeType);
if (s3Url) {
const isImage = mimeType?.startsWith('image/');
atts.push(`[${isImage?'图片':'文件'}: ${filename}](${s3Url}) (必须下载附件，了解内容后才能回复)`);
}
}
} catch (e) { err(` ❌ 附件失败: ${e.message}`); }
}
}
if (atts.length) txt += (txt ? '\n\n' : '') + atts.join('\n');
const isCommand = txt.trim().startsWith('/');
if (txt || atts.length > 0) {
if (!txt && atts.length > 0) txt = "发送了附件。";
await new Promise(r => setTimeout(r, 500));
triggerAgent(cid, m.user_id, sender, txt, isCommand);
}
}
}
} catch (e) {
if(e.message!=="No messages") err(`[${state.name}] 跳过: ${e.message}`);
} finally { state.syncing = false; }
}
async function loop() {
const delay = Math.max(parseInt(POLL_INTERVAL || 5000), 2000);
if (isClawReady) {
for (const cid of channelRuntime.keys()) {
await syncChannel(cid);
await new Promise(r => setTimeout(r, 200));
}
}
setTimeout(loop, delay);
}
loop();