OpenClaw 架构中消息流入中枢 —— monitor-inbox.ts 如何解析、去重与防抖

// monitor-inbox.ts
const seenMessageIds = new Set<string>();

function isDuplicate(message: RawMessage): boolean {
  if (message.id && seenMessageIds.has(message.id)) {
    return true;
  }
  if (message.id) {
    seenMessageIds.add(message.id);
    // 5 分钟后自动清理（防止内存无限增长）
    setTimeout(() => seenMessageIds.delete(message.id), 300_000);
  }
  return false;
}

const recentHashes = new Map<string, number>(); // hash → timestamp

function isFuzzyDuplicate(content: string, sessionKey: string): boolean {
  const hash = md5(`${sessionKey}:${content}`);
  const now = Date.now();
  
  // 清理 2 秒前的记录
  for (const [h, ts] of recentHashes.entries()) {
    if (now - ts > 2000) recentHashes.delete(h);
  }

  if (recentHashes.has(hash)) {
    return true; // 2 秒内相同内容
  }
  
  recentHashes.set(hash, now);
  return false;
}

// monitor-inbox.ts
interface DebounceBuffer {
  messages: string[];
  lastActivity: number;
  timeoutId: NodeJS.Timeout | null;
}

const debounceBuffers = new Map<string, DebounceBuffer>(); // sessionKey → buffer
const DEBOUNCE_WINDOW = 1500; // 1.5 秒

function enqueueMessage(sessionKey: string, text: string) {
  let buf = debounceBuffers.get(sessionKey);
  
  if (!buf) {
    buf = { messages: [], lastActivity: 0, timeoutId: null };
    debounceBuffers.set(sessionKey, buf);
  }

  buf.messages.push(text);
  buf.lastActivity = Date.now();

  // 清除旧定时器
  if (buf.timeoutId) clearTimeout(buf.timeoutId);

  // 设置新定时器
  buf.timeoutId = setTimeout(() => {
    flushBuffer(sessionKey);
    debounceBuffers.delete(sessionKey);
  }, DEBOUNCE_WINDOW);
}

// monitor-inbox.ts
type MediaType = 'image' | 'audio' | 'video' | 'document';

function detectMediaType(mime: string): MediaType | null {
  if (mime.startsWith('image/')) return 'image';
  if (mime.startsWith('audio/')) return 'audio';
  // ...
  return null;
}

async function handleMedia(message: RawMessage) {
  if (!message.mediaUrl) return;

  const mediaType = detectMediaType(message.mimeType);
  if (!mediaType) return;

  // 生成安全路径
  const ext = mime.getExtension(message.mimeType) || 'bin';
  const filename = `${nanoid()}.${ext}`;
  const localPath = path.join(
    process.env.MEDIA_DIR || './media',
    message.sessionKey,
    filename
  );

  // 下载（带超时与重试）
  await downloadWithRetry(message.mediaUrl, localPath, {
    timeout: 30_000,
    retries: 3
  });

  // 替换原始消息中的 URL 为本地路径
  message.localMediaPath = localPath;
}

emitACPEvent('chat.processedMessage', {
  sessionKey: 'wa:+1234567890',
  content: '帮我查一下订单 ORD-123 的状态',
  media: [
    { type: 'image', path: '/media/wa_+1234567890/abc123.jpg' }
  ],
  sourceChannel: 'whatsapp',
  timestamp: 1710234567
});

openclaw_inbox_messages_total{channel="whatsapp", action="deduped"} 12
openclaw_inbox_messages_total{channel="web", action="debounced"} 8
openclaw_inbox_media_downloaded_bytes_total 4567890

[INFO]  Merged 3 messages from wa:+1234567890 → "重启数据库并检查日志"
[DEBUG] Downloaded image to /media/wa_+1234567890/img_a1b2c3.jpg
[WARN]  Duplicate message id=msg_abc123 ignored