openclaw-voc-skill/claude-code/claude-code-qiwe-assistant/scripts/start-relay-client.js

#!/usr/bin/env node
/**
 * Relay 长轮询客户端
 *
 * 独立进程运行，从中央 Relay 拉取属于本租户的加密事件，
 * 用本地 RSA 私钥解密后构造 v1 envelope 并交给 processWebhookEvents 处理。
 *
 * 启动方式：
 *   node scripts/start-relay-client.js [device-guid]
 *   npm run relay
 */

const crypto = require('crypto');
const fs = require('fs');
const { processWebhookEvents } = require('../mcp/src/core/webhook-server');
const { readQiweiGuid } = require('../mcp/src/core/credentials');
const { getProductMode } = require('../mcp/src/core/product-mode');
const { buildContext } = require('../mcp/src/core/shared-gateway');
const { callFmodeWecomGateway } = require('../mcp/src/providers/fmode-wecom-gateway');
const {
  getRelayBaseUrl,
  getTenantApiSecret,
  getRelayPrivateKey,
  getTenantId,
  getRelayDeviceGuid
} = require('../mcp/src/core/relay-config');
const {
  relayLockPath,
  relayStatePath,
  readJson,
  isProcessAlive
} = require('../mcp/src/core/relay-daemon');

const POLL_WAIT_MS = 30000;
const INITIAL_BACKOFF_MS = 1000;
const MAX_BACKOFF_MS = 60000;
const UPSTREAM_RECONNECT_INTERVAL_MS = 5 * 60 * 1000;
const UPSTREAM_RECONNECT_RETRY_MS = 60 * 1000;
const DEVICE_REFRESH_INTERVAL_MS = 60 * 1000;

let lockOwned = false;
let runtimeState = {
  pid: process.pid,
  startedAt: new Date().toISOString(),
  heartbeatAt: new Date().toISOString(),
  lastPollAt: null,
  lastReceivedAt: null,
  lastAckAt: null,
  lastError: '',
  received: 0,
  acked: 0,
  failed: 0
};

let nextUpstreamReconnectAt = 0;
let startupCatchupPending = true;
let nextDeviceRefreshAt = 0;
let activeDeviceGuids = [];

function writeJsonAtomic(filePath, value) {
  const tempPath = `${filePath}.${process.pid}.tmp`;
  fs.writeFileSync(tempPath, JSON.stringify(value, null, 2), 'utf8');
  fs.renameSync(tempPath, filePath);
}

function writeRuntimeState(patch = {}) {
  runtimeState = { ...runtimeState, ...patch, pid: process.pid, heartbeatAt: new Date().toISOString() };
  writeJsonAtomic(relayStatePath(), runtimeState);
}

function acquireLock() {
  const lockPath = relayLockPath();
  for (let attempt = 0; attempt < 2; attempt += 1) {
    try {
      const fd = fs.openSync(lockPath, 'wx');
      fs.writeFileSync(fd, JSON.stringify({ pid: process.pid, startedAt: runtimeState.startedAt }, null, 2));
      fs.closeSync(fd);
      lockOwned = true;
      return true;
    } catch (error) {
      if (error.code !== 'EEXIST') throw error;
      const existing = readJson(lockPath, {});
      if (isProcessAlive(existing.pid)) return false;
      try { fs.rmSync(lockPath, { force: true }); } catch {}
    }
  }
  return false;
}

function releaseLock() {
  if (!lockOwned) return;
  const existing = readJson(relayLockPath(), {});
  if (Number(existing.pid) === process.pid) fs.rmSync(relayLockPath(), { force: true });
  lockOwned = false;
  writeRuntimeState({ running: false, stoppedAt: new Date().toISOString() });
}

function decryptPayload(encryptedPayload, privateKeyPem) {
  const key = crypto.createPrivateKey(privateKeyPem);
  if (String(encryptedPayload).startsWith('v2:')) {
    const envelope = JSON.parse(Buffer.from(String(encryptedPayload).slice(3), 'base64').toString('utf8'));
    const aesKey = crypto.privateDecrypt(
      { key, oaepHash: 'sha256' },
      Buffer.from(envelope.key, 'base64')
    );
    const decipher = crypto.createDecipheriv(
      'aes-256-gcm',
      aesKey,
      Buffer.from(envelope.iv, 'base64')
    );
    decipher.setAuthTag(Buffer.from(envelope.tag, 'base64'));
    return Buffer.concat([
      decipher.update(Buffer.from(envelope.ciphertext, 'base64')),
      decipher.final()
    ]).toString('utf8');
  }
  const buffer = Buffer.from(encryptedPayload, 'base64');
  const decrypted = crypto.privateDecrypt({ key, oaepHash: 'sha256' }, buffer);
  return decrypted.toString('utf8');
}

async function ackEvents(baseUrl, apiSecret, guid, eventIds) {
  if (!eventIds.length) return 0;
  const res = await fetch(`${baseUrl}/api/relay/ack`, {
    method: 'POST',
    headers: {
      'Content-Type': 'application/json',
      Authorization: `Bearer ${apiSecret}`
    },
    body: JSON.stringify({ guid, eventIds })
  });

  if (!res.ok) {
    throw new Error(`ACK failed: ${res.status} ${await res.text()}`);
  }

  const data = await res.json();
  const ackedCount = Number(data.ackedCount) || eventIds.length;
  console.log(`[RelayClient] ACK ${ackedCount} 条事件`);
  return ackedCount;
}

async function runPollOnce(baseUrl, apiSecret, guid, privateKey, options = {}) {
  const trackState = options.trackState !== false;
  const res = await fetch(`${baseUrl}/api/relay/poll`, {
    method: 'POST',
    headers: {
      'Content-Type': 'application/json',
      Authorization: `Bearer ${apiSecret}`
    },
    body: JSON.stringify({ guid, batchSize: 100, waitMs: POLL_WAIT_MS })
  });

  if (!res.ok) {
    throw new Error(`poll failed: ${res.status} ${await res.text()}`);
  }

  const data = await res.json();
  if (trackState) writeRuntimeState({ lastPollAt: new Date().toISOString(), lastError: '' });
  if (!data.events || !data.events.length) return { received: 0, acked: 0, failed: 0 };

  console.log(`[RelayClient] 取回 ${data.events.length} 条事件`);
  const eventIds = [];
  let failed = 0;

  for (const event of data.events) {
    try {
      const decrypted = decryptPayload(event.encryptedPayload, privateKey);
      const payload = JSON.parse(decrypted);
      const envelope = payload && payload.version === '2.0'
        ? { ...payload, source: 'relay', __rawBody: decrypted }
        : payload && typeof payload.code === 'number' && Array.isArray(payload.data)
          ? { ...payload, source: 'relay', __rawBody: decrypted }
          : { code: 0, msg: 'from-relay', data: Array.isArray(payload) ? payload : [payload], source: 'relay', __rawBody: decrypted };
      const processed = await processWebhookEvents(envelope);
      if (processed.errors) throw new Error(`webhook processing failed for ${processed.errors} event(s)`);
      eventIds.push(event.eventId);
    } catch (err) {
      console.error(`[RelayClient] 解密/处理失败 eventId=${event.eventId}:`, err.message);
      failed += 1;
    }
  }

  const acked = await ackEvents(baseUrl, apiSecret, guid, eventIds);
  if (trackState) {
    writeRuntimeState({
      lastReceivedAt: new Date().toISOString(),
      lastAckAt: acked ? new Date().toISOString() : runtimeState.lastAckAt,
      received: runtimeState.received + data.events.length,
      acked: runtimeState.acked + acked,
      failed: runtimeState.failed + failed,
      lastError: failed ? `${failed} event(s) failed and remain pending` : ''
    });
  }
  return { received: data.events.length, acked, failed };
}

function resolveGuid() {
  // 命令行参数 > 环境变量 > relay-config.json > credentials
  return process.argv[2] || process.env.RELAY_DEVICE_GUID || getRelayDeviceGuid() || readQiweiGuid() || '';
}

function resolveRuntimeConfig() {
  return {
    baseUrl: getRelayBaseUrl(),
    apiSecret: getTenantApiSecret(),
    privateKey: getRelayPrivateKey(),
    tenantId: getTenantId(),
    guid: resolveGuid()
  };
}

async function refreshTenantDevices(baseUrl, apiSecret, primaryGuid) {
  if (Date.now() < nextDeviceRefreshAt && activeDeviceGuids.length) return activeDeviceGuids;
  const response = await fetch(`${baseUrl}/api/tenant/status`, {
    headers: { Authorization: `Bearer ${apiSecret}` },
    signal: AbortSignal.timeout(15000)
  });
  if (!response.ok) throw new Error(`device discovery failed: ${response.status} ${await response.text()}`);
  const data = await response.json();
  const discovered = Array.isArray(data.devices)
    ? data.devices.map(device => String(device.guid || '').trim()).filter(Boolean)
    : [];
  activeDeviceGuids = [...new Set([primaryGuid, ...discovered].filter(Boolean))];
  nextDeviceRefreshAt = Date.now() + DEVICE_REFRESH_INTERVAL_MS;
  writeRuntimeState({ activeDeviceGuids });
  return activeDeviceGuids;
}

async function reconnectUpstreamCallback() {
  const ctx = buildContext({});
  const result = await callFmodeWecomGateway({
    gatewayPath: '/relay/connect',
    body: { uid: ctx.uid },
    token: ctx.token,
    apiBase: ctx.apiBase,
    timeoutMs: 60000
  });
  const connected = result.data && result.data.data !== undefined ? result.data.data : result.data;
  if (connected && connected.connected === false) throw new Error('Fmode Relay connection was not accepted');
  nextUpstreamReconnectAt = Date.now() + UPSTREAM_RECONNECT_INTERVAL_MS;
  writeRuntimeState({ lastConnectAt: new Date().toISOString(), lastConnectError: '' });
  console.log('[RelayClient] Fmode Relay 回调已重新连接');
  if (startupCatchupPending) {
    try {
      const { syncConversations } = require('../mcp/src/dashboard/agent-service');
      const catchup = await syncConversations();
      startupCatchupPending = false;
      writeRuntimeState({
        lastCatchupAt: new Date().toISOString(),
        lastCatchupError: '',
        lastCatchupMessage: catchup.assistantMessage || ''
      });
      console.log('[RelayClient] 启动补采完成:', catchup.assistantMessage || 'ok');
    } catch (error) {
      nextUpstreamReconnectAt = Date.now() + UPSTREAM_RECONNECT_RETRY_MS;
      writeRuntimeState({ lastCatchupError: error.message });
      console.warn('[RelayClient] 启动补采失败:', error.message);
    }
  }
  return connected;
}

async function maintainUpstreamCallback() {
  if (Date.now() < nextUpstreamReconnectAt) return;
  try {
    await reconnectUpstreamCallback();
  } catch (error) {
    nextUpstreamReconnectAt = Date.now() + UPSTREAM_RECONNECT_RETRY_MS;
    writeRuntimeState({ lastConnectError: error.message });
    console.warn('[RelayClient] Fmode Relay 回调重连失败:', error.message);
  }
}

async function main() {
  if (!acquireLock()) {
    console.log('[RelayClient] 已有 Relay 消费进程运行，本进程退出');
    return;
  }
  writeRuntimeState({ running: true });
  const product = getProductMode();
  if (product.mode !== 'enterprise') {
    console.error('[RelayClient] 当前为个人版，请使用工作台本地监听；企业 Relay Client 未启动');
    process.exit(1);
  }
  let config = resolveRuntimeConfig();

  if (!config.apiSecret || !config.privateKey || !config.tenantId) {
    console.error('[RelayClient] 缺少配置：请检查 .env.local 中的 TENANT_API_SECRET、RELAY_PRIVATE_KEY、TENANT_ID');
    process.exit(1);
  }
  if (!config.guid) {
    console.error('[RelayClient] 缺少 deviceGuid：请通过命令行传入，或配置 RELAY_DEVICE_GUID / relay-config.json / 完成企微登录');
    process.exit(1);
  }

  console.log(`[RelayClient] 启动 Relay 轮询: ${config.baseUrl}`);
  console.log(`[RelayClient] tenantId=${config.tenantId}, guid=${config.guid}`);

  let backoff = INITIAL_BACKOFF_MS;
  let configIdentity = `${config.baseUrl}|${config.tenantId}|${config.guid}`;

  while (true) {
    try {
      const latestConfig = resolveRuntimeConfig();
      if (!latestConfig.apiSecret || !latestConfig.privateKey || !latestConfig.tenantId || !latestConfig.guid) {
        throw new Error('Relay runtime configuration became incomplete');
      }
      const latestIdentity = `${latestConfig.baseUrl}|${latestConfig.tenantId}|${latestConfig.guid}`;
      if (latestIdentity !== configIdentity) {
        config = latestConfig;
        configIdentity = latestIdentity;
        nextUpstreamReconnectAt = 0;
        nextDeviceRefreshAt = 0;
        activeDeviceGuids = [];
        startupCatchupPending = true;
        writeRuntimeState({
          tenantId: config.tenantId,
          deviceGuid: config.guid,
          lastConfigReloadAt: new Date().toISOString()
        });
        console.log(`[RelayClient] 已切换 Relay 配置: tenantId=${config.tenantId}, guid=${config.guid}`);
      } else {
        config = latestConfig;
      }
      await maintainUpstreamCallback();
      let deviceGuids;
      try {
        deviceGuids = await refreshTenantDevices(config.baseUrl, config.apiSecret, config.guid);
      } catch (error) {
        deviceGuids = activeDeviceGuids.length ? activeDeviceGuids : [config.guid];
        nextDeviceRefreshAt = Date.now() + Math.min(UPSTREAM_RECONNECT_RETRY_MS, DEVICE_REFRESH_INTERVAL_MS);
        console.warn('[RelayClient] 租户设备发现失败，继续轮询已知设备:', error.message);
      }
      const settled = await Promise.allSettled(
        deviceGuids.map(guid => runPollOnce(config.baseUrl, config.apiSecret, guid, config.privateKey, { trackState: false }))
      );
      const results = settled.filter(item => item.status === 'fulfilled').map(item => item.value);
      const errors = settled.filter(item => item.status === 'rejected').map(item => item.reason?.message || String(item.reason));
      if (!results.length && errors.length) throw new Error(errors.join('; '));
      const received = results.reduce((sum, item) => sum + item.received, 0);
      const acked = results.reduce((sum, item) => sum + item.acked, 0);
      const failed = results.reduce((sum, item) => sum + item.failed, 0);
      writeRuntimeState({
        lastPollAt: new Date().toISOString(),
        lastReceivedAt: received ? new Date().toISOString() : runtimeState.lastReceivedAt,
        lastAckAt: acked ? new Date().toISOString() : runtimeState.lastAckAt,
        received: runtimeState.received + received,
        acked: runtimeState.acked + acked,
        failed: runtimeState.failed + failed,
        lastError: [...errors, ...(failed ? [`${failed} event(s) failed and remain pending`] : [])].join('; ')
      });
      backoff = INITIAL_BACKOFF_MS;
    } catch (err) {
      console.error('[RelayClient] 轮询异常:', err.message);
      writeRuntimeState({ lastError: err.message });
      console.log(`[RelayClient] ${backoff}ms 后重试...`);
      await new Promise((resolve) => setTimeout(resolve, backoff));
      backoff = Math.min(backoff * 2, MAX_BACKOFF_MS);
    }
  }
}

if (require.main === module) {
  process.once('SIGINT', () => { releaseLock(); process.exit(0); });
  process.once('SIGTERM', () => { releaseLock(); process.exit(0); });
  process.once('exit', releaseLock);
  main().catch((err) => {
    console.error('[RelayClient] 致命错误:', err);
    writeRuntimeState({ running: false, lastError: err.message });
    releaseLock();
    process.exit(1);
  });
}

module.exports = { decryptPayload, ackEvents, runPollOnce, resolveGuid, resolveRuntimeConfig, refreshTenantDevices, reconnectUpstreamCallback, maintainUpstreamCallback, acquireLock, releaseLock };