openclaw-voc-skill/claude-code/claude-code-qiwe-assistant/scripts/message-polling-worker.js

#!/usr/bin/env node
/**
 * Message polling fallback worker.
 *
 * This is a demo-safe alternative when webhook callback configuration cannot
 * be changed. It pulls messages through /msg/syncMsg and writes them into the
 * same outputs/messages tree used by webhook processing.
 */

const fs = require('fs');
const path = require('path');
const { outputsRoot } = require('../mcp/src/core/output-paths');
const { buildContext, gatewayCall } = require('../mcp/src/core/shared-gateway');
const { readQiweiGuid } = require('../mcp/src/core/credentials');
const { recordGroupSync } = require('../mcp/src/core/dashboard-state');

const INTERVAL_MS = Math.max(3000, Number(process.env.MESSAGE_POLL_INTERVAL_MS || 5000));
const PAGE_LIMIT = Math.max(20, Math.min(500, Number(process.env.MESSAGE_POLL_PAGE_LIMIT || 200)));
const MAX_PAGES_PER_TICK = Math.max(1, Math.min(20, Number(process.env.MESSAGE_POLL_MAX_PAGES || 5)));
const IMPORT_HISTORY = process.argv.includes('--import-history') || process.env.MESSAGE_POLL_IMPORT_HISTORY === 'true';
const RESET = process.argv.includes('--reset');
const ONCE = process.argv.includes('--once');
const LATEST_SEQ_PROBE = 999999999;
const REQUEST_RETRIES = Math.max(0, Math.min(5, Number(process.env.MESSAGE_POLL_REQUEST_RETRIES || 3)));
let gbkEncodeMap = null;

function messagesDir() {
  return path.join(outputsRoot(), 'messages');
}

function statePath() {
  return path.join(messagesDir(), 'polling-state.json');
}

function ensureDir(dir) {
  fs.mkdirSync(dir, { recursive: true });
}

function safeReadJson(filePath, fallback = null) {
  try {
    if (!fs.existsSync(filePath)) return fallback;
    return JSON.parse(fs.readFileSync(filePath, 'utf8').replace(/^\uFEFF/, ''));
  } catch {
    return fallback;
  }
}

function writeJson(filePath, data) {
  ensureDir(path.dirname(filePath));
  fs.writeFileSync(filePath, JSON.stringify(data, null, 2), 'utf8');
}

function sleep(ms) {
  return new Promise(resolve => setTimeout(resolve, ms));
}

function readState() {
  if (RESET) return {};
  return safeReadJson(statePath(), {}) || {};
}

function writeState(state) {
  writeJson(statePath(), { ...state, updatedAt: new Date().toISOString() });
}

function sanitizeFilePart(value, fallback = 'message') {
  const safe = String(value || '').replace(/[^a-zA-Z0-9_-]/g, '_').slice(0, 90);
  return safe || fallback;
}

function conversationIdForMessage(msg) {
  const roomId = String(msg.fromRoomId || '');
  if (roomId && roomId !== '0') return roomId;

  const senderId = String(msg.senderId || '');
  const receiverId = String(msg.receiverId || '');
  const otherId = senderId && senderId !== readQiweiGuid() ? senderId : receiverId || senderId;
  return `private-${sanitizeFilePart(otherId || 'unknown')}`;
}

function getGbkEncodeMap() {
  if (gbkEncodeMap) return gbkEncodeMap;
  gbkEncodeMap = new Map();
  const decoder = new TextDecoder('gbk');
  for (let first = 0x81; first <= 0xfe; first++) {
    for (let second = 0x40; second <= 0xfe; second++) {
      if (second === 0x7f) continue;
      const bytes = Buffer.from([first, second]);
      const char = decoder.decode(bytes);
      if (!char || char === '\uFFFD' || char.length !== 1) continue;
      if (!gbkEncodeMap.has(char)) gbkEncodeMap.set(char, bytes);
    }
  }
  return gbkEncodeMap;
}

function repairUtf8DecodedAsGbk(value) {
  const text = String(value || '');
  if (!/[\u4e00-\u9fa5]/.test(text)) return text;
  const map = getGbkEncodeMap();
  const chunks = [];
  for (const char of text) {
    const code = char.codePointAt(0);
    if (code <= 0x7f) {
      chunks.push(Buffer.from([code]));
      continue;
    }
    const bytes = map.get(char);
    if (!bytes) return text;
    chunks.push(bytes);
  }
  const repaired = Buffer.concat(chunks).toString('utf8');
  if (repaired.includes('\uFFFD')) return text;
  return /[\u4e00-\u9fa5]/.test(repaired) ? repaired : text;
}

function cleanExtractedText(value) {
  const cleaned = String(value || '').replace(/[\u0000-\u001f\u007f]/g, '').trim();
  return repairUtf8DecodedAsGbk(cleaned);
}

function extractUtf8TextFromBase64(value) {
  if (!value) return '';
  let buffer;
  try {
    buffer = Buffer.from(String(value), 'base64');
  } catch {
    return '';
  }
  const candidates = [];
  for (let i = 0; i < buffer.length - 1; i++) {
    const len = buffer[i];
    if (!len || len > 240 || i + 1 + len > buffer.length) continue;
    const text = cleanExtractedText(buffer.slice(i + 1, i + 1 + len).toString('utf8'));
    if (!text || text.includes('\uFFFD')) continue;
    if (/[\u4e00-\u9fa5]/.test(text)) candidates.push(text);
  }
  return candidates.sort((a, b) => b.length - a.length)[0] || '';
}

function extractContent(msg) {
  const rawText = extractUtf8TextFromBase64(msg.base64RawData || msg.msgData?.extras?.base64RawData);
  if (rawText) return rawText;
  if (typeof msg.content === 'string' && msg.content) return cleanExtractedText(msg.content);
  if (typeof msg.msgContent === 'string' && msg.msgContent) return cleanExtractedText(msg.msgContent);
  if (msg.msgData) {
    if (typeof msg.msgData.content === 'string') return cleanExtractedText(msg.msgData.content);
    if (Array.isArray(msg.msgData.moreDetail)) {
      return cleanExtractedText(msg.msgData.moreDetail.map(item => item && item.text).filter(Boolean).join(''));
    }
    if (typeof msg.msgData.notifyTitle === 'string') return cleanExtractedText(msg.msgData.notifyTitle);
  }
  return '';
}

function normalizeMessage(msg, conversationId) {
  const timestamp = Number(msg.timestamp) || 0;
  return {
    msgId: String(msg.msgUniqueIdentifier || msg.msgServerId || `${conversationId}_${msg.seq || Date.now()}`),
    seq: Number(msg.seq) || 0,
    senderId: String(msg.senderId || ''),
    senderName: String(msg.senderName || ''),
    receiverId: String(msg.receiverId || ''),
    msgType: String(msg.msgType || ''),
    content: extractContent(msg),
    timestamp: timestamp ? new Date(timestamp * 1000).toISOString() : new Date().toISOString(),
    isRevoked: Boolean(msg.isRevoked),
    source: 'polling',
    rawData: msg
  };
}

function messageFilePath(conversationId, message) {
  const dir = path.join(messagesDir(), sanitizeFilePart(conversationId, 'conversation'));
  ensureDir(dir);
  const seq = Number(message.seq) || 0;
  return path.join(dir, `${seq}-${sanitizeFilePart(message.msgId)}.json`);
}

function saveMessage(msg) {
  const conversationId = conversationIdForMessage(msg);
  const message = normalizeMessage(msg, conversationId);
  const filePath = messageFilePath(conversationId, message);
  if (fs.existsSync(filePath)) return { written: false, conversationId, filePath, message };
  writeJson(filePath, message);
  return { written: true, conversationId, filePath, message };
}

async function syncPage(ctx, msgSeq) {
  let lastError = null;
  for (let attempt = 0; attempt <= REQUEST_RETRIES; attempt++) {
    try {
      const data = await gatewayCall(ctx, '/msg/syncMsg', {
        guid: ctx.guid,
        msgSeq,
        limit: PAGE_LIMIT
      });
      return {
        messages: Array.isArray(data && data.syncMsgList) ? data.syncMsgList : [],
        hasMore: Boolean(data && data.hasMore),
        nextSeq: data && data.travelSyncKey !== undefined ? Number(data.travelSyncKey) : msgSeq
      };
    } catch (err) {
      lastError = err;
      if (attempt >= REQUEST_RETRIES) break;
      await sleep(500 * (attempt + 1));
    }
  }
  throw lastError;
}

async function readLatestSessionSeq(ctx) {
  const data = await gatewayCall(ctx, '/session/getSessionPage', {
    guid: ctx.guid,
    page: 1,
    limit: 20
  });
  const seq = Number(data && data.currentSeq);
  return Number.isFinite(seq) && seq > 0 ? seq : 0;
}

async function seedCursor(ctx) {
  let msgSeq = 0;
  try {
    msgSeq = await readLatestSessionSeq(ctx);
  } catch {
    const page = await syncPage(ctx, LATEST_SEQ_PROBE);
    msgSeq = page.nextSeq || LATEST_SEQ_PROBE;
  }
  writeState({ msgSeq, seededAt: new Date().toISOString(), imported: false });
  console.log(`[MessagePolling] Seeded cursor at ${msgSeq}. New messages after this point will be imported.`);
  return { msgSeq };
}

async function runOnce(ctx, state) {
  let msgSeq = Number(state.msgSeq || 0);
  let pages = 0;
  let fetched = 0;
  let written = 0;
  const touched = new Map();

  while (pages < MAX_PAGES_PER_TICK) {
    pages++;
    const page = await syncPage(ctx, msgSeq);
    fetched += page.messages.length;
    msgSeq = page.nextSeq || msgSeq + 1;

    for (const msg of page.messages) {
      const saved = saveMessage(msg);
      if (!saved.written) continue;
      written++;
      const current = touched.get(saved.conversationId) || { messageCount: 0, lastMsgAt: null, lastSyncSeq: 0 };
      current.messageCount++;
      current.lastMsgAt = saved.message.timestamp;
      current.lastSyncSeq = Math.max(current.lastSyncSeq || 0, Number(saved.message.seq) || 0);
      touched.set(saved.conversationId, current);
    }

    if (!page.hasMore || !page.messages.length) break;
  }

  for (const [conversationId, info] of touched.entries()) {
    recordGroupSync(conversationId, info);
  }

  const nextState = {
    ...state,
    msgSeq,
    lastFetched: fetched,
    lastWritten: written,
    lastRunAt: new Date().toISOString()
  };
  writeState(nextState);
  return nextState;
}

async function main() {
  const guid = String(process.argv.find(arg => arg.startsWith('--guid='))?.slice('--guid='.length) || readQiweiGuid() || '').trim();
  if (!guid) {
    console.error('[MessagePolling] Missing guid. Log in first or pass --guid=<device-guid>.');
    process.exit(1);
  }

  const ctx = buildContext({ guid });
  console.log(`[MessagePolling] Started. guid=${guid}, interval=${INTERVAL_MS}ms`);

  let state = readState();
  if (!state.msgSeq && !IMPORT_HISTORY) {
    state = await seedCursor(ctx);
  }

  while (true) {
    try {
      state = await runOnce(ctx, state);
      console.log(`[MessagePolling] fetched=${state.lastFetched}, written=${state.lastWritten}, nextSeq=${state.msgSeq}`);
    } catch (err) {
      console.error('[MessagePolling] Poll failed:', err.message);
    }
    if (ONCE) break;
    await new Promise(resolve => setTimeout(resolve, INTERVAL_MS));
  }
}

main().catch(err => {
  console.error('[MessagePolling] Fatal:', err);
  process.exit(1);
});