openclaw-wx-skills/channel-plugin/src/monitor.ts

/**
 * Inbound monitor — polls the wechat-agent backend and dispatches new
 * messages through the OpenClaw agent reply pipeline.
 *
 * Dispatch sequence mirrors @openclaw/msteams and @openclaw/zalo:
 *   1. resolveAgentRoute      → session key + agent id
 *   2. activity.record        → inbound counter
 *   3. enqueueSystemEvent     → heartbeat preview
 *   4. formatInboundEnvelope  → wrapped body
 *   5. finalizeInboundContext → ctx payload
 *   6. updateLastRoute        → stick session to this channel
 *   7. createReplyDispatcherWithTyping + deliver callback
 *   8. withReplyDispatcher → dispatchReplyFromConfig
 *
 * FAQ fast-path (D1=B): rules run before step 1 and can short-circuit
 * the entire pipeline by sending a canned reply directly via sendText.
 *
 * Batch window (D3=B): messages are accumulated per-conversation for
 * batchWindowMs and dispatched as one agent turn.
 */

import type {
  ChannelAccountSnapshot,
  OpenClawConfig,
  ReplyPayload,
  RuntimeEnv,
} from "openclaw/plugin-sdk";
import { DEFAULT_ACCOUNT_ID, createReplyPrefixOptions } from "./sdk.js";
import { getRuntime } from "./runtime.js";
import {
  createClient,
  type WechatAgentClient,
} from "./client.js";
import { matchFaq, shouldAlsoDispatchAgent } from "./faq.js";
import { CooldownManager, mergeBatchText } from "./cooldown.js";
import {
  CHANNEL_ID,
  getChannelCfg,
  normalizeAllowFrom,
  resolveApiBase,
} from "./config.js";
import type { Logger, RawWechatMessage, WechatAgentConfig } from "./types.js";
import {
  advanceCursor,
  compareMsgId,
  isAfterCursor,
  loadCursor,
  saveCursor,
  toEpochMs,
  type InboundCursor,
} from "./cursor-store.js";
import { join } from "node:path";
import { withRetry } from "./retry.js";

export type MonitorOpts = {
  cfg: OpenClawConfig;
  runtime?: RuntimeEnv;
  abortSignal?: AbortSignal;
  /**
   * Shallow-merge patch into the ChannelAccountSnapshot. Provided by the
   * caller (gateway.startAccount) which wraps ctx.getStatus+ctx.setStatus.
   */
  patchStatus: (patch: Partial<ChannelAccountSnapshot>) => void;
  accountId: string;
};

export type MonitorResult = {
  shutdown: () => Promise<void>;
};

const POLL_MIN_INTERVAL_MS = 3000;
const SEEN_MAX = 2000;
const SEEN_TRIM_TO = 1000;
const DEFAULT_GROUP_KEYWORDS = ["帮我", "请问"];
const DEFAULT_GROUP_MENTION_TOKENS = ["@bot", "@助手"];

export async function startMonitor(opts: MonitorOpts): Promise<MonitorResult> {
  const channelCfg = getChannelCfg(opts.cfg);
  const log = makeLogger(opts.runtime);

  if (!channelCfg) {
    log.info("no channel config; poller not started");
    return noopResult();
  }
  if (channelCfg.enabled === false) {
    log.info("channel disabled; poller not started");
    return noopResult();
  }
  if (!channelCfg.apiBase) {
    log.warn(`channels.${CHANNEL_ID}.apiBase is missing; poller not started`);
    return noopResult();
  }

  const apiBase = resolveApiBase(channelCfg);
  const client = createClient(apiBase);
  const pollIntervalMs = Math.max(
    POLL_MIN_INTERVAL_MS,
    channelCfg.pollIntervalMs ?? 10000
  );
  const batchWindowMs = Math.max(0, channelCfg.batchWindowMs ?? 3000);
  const replyCooldownMs = Math.max(0, (channelCfg.replyCooldownSecPerWxid ?? 5) * 1000);

  // Persistent inbound cursor: best-effort resume after restarts.
  // We store it alongside OpenClaw's session store so admins can back it up
  // together with other channel state.
  const core = getRuntime();
  const sessionCfg = (opts.cfg as Record<string, unknown>).session as
    | { store?: string }
    | undefined;
  const cursorStoreDir = core.channel.session.resolveStorePath(sessionCfg?.store, {
    agentId: CHANNEL_ID,
  });
  const cursorPath = join(cursorStoreDir, `${CHANNEL_ID}.inbound-cursor.json`);
  let cursor: InboundCursor | null = await loadCursor(cursorPath);

  // Resolve bot's own wxid so we can filter self-echoes.
  let selfWxid: string | null = channelCfg.selfWxid?.trim() || null;
  if (!selfWxid) {
    try {
      selfWxid = await client.getSelfWxid();
      if (selfWxid) log.info(`detected self wxid: ${selfWxid}`);
    } catch {
      /* ignore — backend may not expose it */
    }
  }

  log.info(
    `starting poller apiBase=${apiBase} interval=${pollIntervalMs}ms batch=${batchWindowMs}ms cooldown=${replyCooldownMs}ms selfWxid=${selfWxid ?? "<auto>"}`
  );
  opts.patchStatus({
    running: true,
    lastStartAt: Date.now(),
    lastError: null,
  });

  // Local counters: surfaced via logs only. ChannelAccountSnapshot does not
  // carry channel-specific metrics, so we keep them out of the snapshot to
  // stay schema-clean.
  const counters = {
    pollsOk: 0,
    pollsFail: 0,
    inboundCount: 0,
    outboundCount: 0,
    faqHits: 0,
  };

  // Message dedupe window
  const seenMsgIds = new Set<string>();
  const markSeen = (id: string): boolean => {
    if (seenMsgIds.has(id)) return false;
    seenMsgIds.add(id);
    if (seenMsgIds.size > SEEN_MAX) {
      const overflow = seenMsgIds.size - SEEN_TRIM_TO;
      const iter = seenMsgIds.values();
      for (let i = 0; i < overflow; i++) {
        const next = iter.next();
        if (next.done) break;
        seenMsgIds.delete(next.value);
      }
    }
    return true;
  };

  // Batch + cooldown per wxid
  const cooldown = new CooldownManager({
    batchWindowMs,
    replyCooldownMs,
    onFlush: async (batch) => {
      try {
        await dispatchBatchToAgent({
          batch,
          cfg: opts.cfg,
          channelCfg,
          client,
          log,
          onOutboundSent: () => {
            counters.outboundCount++;
            opts.patchStatus({ lastOutboundAt: Date.now() });
            if (batch[0]) cooldown.noteReply(resolveConversationId(batch[0]));
          },
        });
      } catch (err) {
        log.error(`dispatch error: ${String(err)}`);
      }
    },
  });

  // Polling task
  let pollInFlight = false;
  const tick = async (): Promise<void> => {
    if (pollInFlight) return;
    pollInFlight = true;
    try {
      const online = await client.checkOnline();
      opts.patchStatus({ connected: online });
      if (!online) {
        counters.pollsFail++;
        log.debug(`poll fail (offline) pollsFail=${counters.pollsFail}`);
        return;
      }

      const messages = await client.getMessages({
        direction: "received",
        limit: 50,
        signal: opts.abortSignal,
      });
      counters.pollsOk++;

      // Always process in stable order so agent context is consistent.
      messages.sort((a, b) => {
        const ta = toEpochMs(a.timestamp);
        const tb = toEpochMs(b.timestamp);
        if (ta !== tb) return ta - tb;
        return compareMsgId(String(a.msgId), String(b.msgId));
      });

      for (const m of messages) {
        // Cursor gate first: don't let old messages fill the dedupe window.
        if (!isAfterCursor(m, cursor)) continue;
        if (!markSeen(m.msgId)) continue;
        const ignoredReason = shouldIgnoreMessage(m, channelCfg, selfWxid);
        if (ignoredReason) {
          log.debug(`message dropped ${m.msgId}: ${ignoredReason}`);
          continue;
        }
        if (!m.isGroup && !passesDmPolicy(m, channelCfg)) {
          log.debug(`dm policy drop ${m.fromWxid}`);
          continue;
        }
        if (m.isGroup && !isGroupTriggered(m, channelCfg, selfWxid)) {
          log.debug(`group not triggered room=${resolveConversationId(m)} msgId=${m.msgId}`);
          continue;
        }
        counters.inboundCount++;
        opts.patchStatus({ lastInboundAt: Date.now() });

        // FAQ fast-path
        const match = matchFaq(m.content, channelCfg.faq);
        if (match) {
          counters.faqHits++;
          log.info(
            `faq hit id=${match.rule.id ?? "-"} keyword=${match.matchedKeyword} hits=${counters.faqHits}`
          );
          await sendFaqReply(client, m, match.rule.reply, log, channelCfg);
          cooldown.noteReply(resolveConversationId(m));
          counters.outboundCount++;
          opts.patchStatus({ lastOutboundAt: Date.now() });

          if (!shouldAlsoDispatchAgent(match)) {
            continue;
          }
          // stop=false: also forward to agent
        }

        // Batch + dispatch
        cooldown.enqueue(m, resolveConversationId(m));

        // Advance cursor after enqueue. The actual agent dispatch is async,
        // but enqueueing is the "accepted for processing" point for this
        // channel. We still keep a bounded dedupe window to tolerate crashes
        // between enqueue and outbound delivery.
        cursor = advanceCursor(cursor, m);
      }

      // Best-effort persist cursor once per tick.
      if (cursor) {
        await saveCursor(cursorPath, cursor);
      }
    } catch (err) {
      counters.pollsFail++;
      const msg = err instanceof Error ? err.message : String(err);
      opts.patchStatus({ lastError: msg });
      log.warn(`poll error: ${msg}`);
    } finally {
      pollInFlight = false;
    }
  };

  const timer = setInterval(() => {
    tick().catch((err) => log.error(`tick crashed: ${String(err)}`));
  }, pollIntervalMs);
  // DO NOT `timer.unref()`. The poller IS the channel's reason to exist;
  // we want it to keep the Node event loop alive as long as the channel
  // is supposed to be running. Without the ref, after any period where
  // the agent dispatch queue drains (e.g. no new inbound, FAQ pipeline
  // idle), Node exits the child process and OpenClaw's health-monitor
  // logs `restarting (reason: stopped)` and bumps `auto-restart
  // attempt N/10`. Shutdown still works cleanly because `shutdown()`
  // calls `clearInterval(timer)` which removes the ref.

  // Kick off first poll immediately (async, no await so startAccount returns).
  tick().catch((err) => log.error(`first tick crashed: ${String(err)}`));

  let shuttingDown = false;
  const shutdown = async (): Promise<void> => {
    if (shuttingDown) return;
    shuttingDown = true;
    log.info(
      `shutting down poller (pollsOk=${counters.pollsOk} pollsFail=${counters.pollsFail} inbound=${counters.inboundCount} outbound=${counters.outboundCount} faqHits=${counters.faqHits})`
    );
    clearInterval(timer);
    cooldown.stop();
    opts.patchStatus({ running: false, lastStopAt: Date.now() });
  };

  if (opts.abortSignal) {
    if (opts.abortSignal.aborted) {
      await shutdown();
    } else {
      opts.abortSignal.addEventListener(
        "abort",
        () => {
          void shutdown();
        },
        { once: true }
      );
    }
  }

  return { shutdown };
}

// ----------------------------- dispatch helpers ----------------------------

type DispatchParams = {
  batch: RawWechatMessage[];
  cfg: OpenClawConfig;
  channelCfg: WechatAgentConfig;
  client: WechatAgentClient;
  log: Logger;
  onOutboundSent: () => void;
};

async function dispatchBatchToAgent(params: DispatchParams): Promise<void> {
  const { batch, cfg, client, log, onOutboundSent, channelCfg } = params;
  if (batch.length === 0) return;

  const core = getRuntime();
  const accountId = DEFAULT_ACCOUNT_ID;
  const first = batch[0]!;
  const last = batch[batch.length - 1]!;
  const isGroup = first.isGroup;
  const conversationId = resolveConversationId(first);
  const senderWxid = resolveSenderWxid(first);
  const senderName = first.fromName?.trim() || senderWxid;
  const mergedText = mergeBatchText(batch);
  const conversationLabel = isGroup
    ? `${conversationId} · ${senderName}`
    : senderName === conversationId
      ? conversationId
      : `${senderName} (${conversationId})`;

  // 1. Route resolution
  const route = core.channel.routing.resolveAgentRoute({
    cfg,
    channel: CHANNEL_ID,
    accountId,
    peer: { kind: isGroup ? "group" : "direct", id: conversationId },
  });
  const sessionKey = route.sessionKey;

  // 2. Record activity
  core.channel.activity.record({
    channel: CHANNEL_ID,
    accountId,
    direction: "inbound",
  });

  // 3. System event (heartbeat preview)
  const preview = mergedText.replace(/\s+/g, " ").slice(0, 160);
  const countSuffix = batch.length > 1 ? ` (×${batch.length})` : "";
  core.system.enqueueSystemEvent(
    `${isGroup ? "WeChat group" : "WeChat DM"} from ${senderName}${countSuffix}: ${preview}`,
    {
      sessionKey,
      contextKey: `${CHANNEL_ID}:message:${conversationId}:${last.msgId}`,
    }
  );

  // 4. Format envelope
  const body = core.channel.reply.formatInboundEnvelope({
    channel: "WeChat",
    from: conversationLabel,
    body: mergedText,
    chatType: isGroup ? "group" : "direct",
    sender: { name: senderName, id: senderWxid },
  });

  // 5. Finalize context
  const ctxPayload = core.channel.reply.finalizeInboundContext({
    Body: body,
    BodyForAgent: mergedText,
    RawBody: mergedText,
    From: `${CHANNEL_ID}:${conversationId}`,
    To: conversationId,
    SessionKey: sessionKey,
    AccountId: route.accountId,
    ChatType: isGroup ? "group" : "direct",
    ConversationLabel: conversationLabel,
    SenderName: senderName,
    SenderId: senderWxid,
    Provider: CHANNEL_ID,
    Surface: CHANNEL_ID,
    MessageSid: last.msgId,
    OriginatingChannel: CHANNEL_ID,
    OriginatingTo: conversationId,
  });

  // 6. Sticky route for DMs
  const sessionCfg = (cfg as Record<string, unknown>).session as
    | { store?: string }
    | undefined;
  const storePath = core.channel.session.resolveStorePath(sessionCfg?.store, {
    agentId: route.agentId,
  });
  await core.channel.session.updateLastRoute({
    storePath,
    sessionKey: route.mainSessionKey,
    deliveryContext: {
      channel: CHANNEL_ID,
        to: conversationId,
      accountId: route.accountId,
    },
  });

  // 7. Reply dispatcher with deliver callback
  const textLimit = core.channel.text.resolveTextChunkLimit(
    cfg,
    CHANNEL_ID,
    accountId,
    { fallbackLimit: 4000 }
  );

  const { onModelSelected, ...prefixOptions } = createReplyPrefixOptions({
    cfg,
    agentId: route.agentId,
    channel: CHANNEL_ID,
    accountId,
  });

  const { dispatcher, replyOptions, markDispatchIdle } =
    core.channel.reply.createReplyDispatcherWithTyping({
      ...prefixOptions,
      humanDelay: core.channel.reply.resolveHumanDelayConfig(cfg, route.agentId),
      deliver: async (payload: ReplyPayload) => {
        const text = payload.text ?? "";
        if (!text.trim()) return;

        const chunkMode = core.channel.text.resolveChunkMode(
          cfg,
          CHANNEL_ID,
          accountId
        );
        const chunks = core.channel.text.chunkMarkdownTextWithMode(
          text,
          textLimit,
          chunkMode
        );
        const parts = chunks.length > 0 ? chunks : [text];

        for (const chunk of parts) {
          if (!chunk) continue;
          const result = await withRetry(
            () =>
              client.sendText({
                toWxid: conversationId,
                content: chunk,
                ats:
                  isGroup && channelCfg.groupReplyAtSender && senderWxid ? senderWxid : "",
              }),
            {
              tries: 4,
              baseDelayMs: 500,
              maxDelayMs: 4000,
              isRetryable: (err) => {
                const msg = err instanceof Error ? err.message : String(err);
                if (/timeout|aborted|ECONN|ENOTFOUND|EAI_AGAIN/i.test(msg)) return true;
                if (/HTTP 5\d\d/i.test(msg)) return true;
                return false;
              },
            }
          );
          if (!result.ok) {
            log.error(`send-text failed ret=${String(result.ret)} msg=${result.msg ?? ""}`);
            throw new Error(
              `send-text failed: ${result.msg ?? `ret=${String(result.ret)}`}`
            );
          }
        }
        log.info(`delivered reply to ${conversationId} (${text.length} chars)`);
        onOutboundSent();
      },
      onError: (err: unknown, info: { kind: string }) => {
        log.error(`reply delivery failed (${info.kind}): ${String(err)}`);
      },
    });

  // 8. Run the LLM pipeline
  await core.channel.reply.withReplyDispatcher({
    dispatcher,
    onSettled: () => markDispatchIdle(),
    run: () =>
      core.channel.reply.dispatchReplyFromConfig({
        ctx: ctxPayload,
        cfg,
        dispatcher,
        replyOptions: {
          ...replyOptions,
          onModelSelected,
        },
      }),
  });
}

async function sendFaqReply(
  client: WechatAgentClient,
  m: RawWechatMessage,
  reply: string,
  log: Logger,
  cfg: WechatAgentConfig
): Promise<void> {
  const conversationId = resolveConversationId(m);
  const senderWxid = resolveSenderWxid(m);
  const result = await client.sendText({
    toWxid: conversationId,
    content: reply,
    ats: m.isGroup && cfg.groupReplyAtSender && senderWxid ? senderWxid : "",
  });
  if (result.ok) {
    log.info(`faq reply sent to ${conversationId}`);
  } else {
    log.warn(
      `faq reply failed to ${conversationId}: ret=${String(result.ret)} msg=${result.msg ?? ""}`
    );
  }
}

// ----------------------------- filters ----------------------------

function shouldIgnoreMessage(
  m: RawWechatMessage,
  cfg: WechatAgentConfig,
  selfWxid: string | null
): string | null {
  // Self-echoes
  if (selfWxid && m.fromWxid === selfWxid) return "self-echo";
  if (selfWxid && m.chatroomMemberWxid === selfWxid) return "self-group-echo";

  // Group switch
  if (m.isGroup && cfg.groupEnabled === false) return "group-disabled";

  // Type filter
  const ignoreTypes = cfg.ignoreMessageTypes ?? [];
  if (ignoreTypes.includes(m.type)) return `ignored-type:${m.type}`;

  // v0.2: inbound still text-only
  if (m.type !== "text") return "non-text";

  // Wxid prefix filter (official accounts etc.)
  const prefixes = cfg.ignoreWxidPrefixes ?? [];
  for (const p of prefixes) {
    if (p && m.fromWxid.startsWith(p)) return `ignored-prefix:${p}`;
  }

  // Exact wxid filter (system contacts)
  const exacts = cfg.ignoreWxidExact ?? [];
  if (exacts.includes(m.fromWxid)) return `ignored-wxid:${m.fromWxid}`;

  // Empty content
  if (!m.content?.trim()) return "empty-content";

  return null;
}

function passesDmPolicy(m: RawWechatMessage, cfg: WechatAgentConfig): boolean {
  const dmPolicy = cfg.dmPolicy ?? "open";
  if (dmPolicy === "open") return true;

  const allowFrom = normalizeAllowFrom(cfg.allowFrom);
  if (allowFrom.includes(m.fromWxid.toLowerCase())) return true;

  // pairing is handled upstream in core; here we treat it like allowlist.
  return false;
}

function resolveConversationId(m: RawWechatMessage): string {
  if (!m.isGroup) return m.fromWxid;
  if (m.chatroomId?.trim()) return m.chatroomId.trim();
  if (m.fromWxid.endsWith("@chatroom")) return m.fromWxid;
  if (m.toWxid.endsWith("@chatroom")) return m.toWxid;
  return m.fromWxid;
}

function resolveSenderWxid(m: RawWechatMessage): string {
  if (!m.isGroup) return m.fromWxid;
  return m.chatroomMemberWxid?.trim() || m.fromWxid;
}

function isGroupTriggered(
  m: RawWechatMessage,
  cfg: WechatAgentConfig,
  selfWxid: string | null
): boolean {
  if (!m.isGroup) return true;
  const content = m.content.toLowerCase();
  const keywords = (cfg.groupKeywords?.length ? cfg.groupKeywords : DEFAULT_GROUP_KEYWORDS)
    .map((s) => s.trim().toLowerCase())
    .filter(Boolean);
  if (keywords.some((kw) => content.includes(kw))) return true;

  const mentionTokens = (
    cfg.groupMentionTokens?.length ? cfg.groupMentionTokens : DEFAULT_GROUP_MENTION_TOKENS
  )
    .map((s) => s.trim().toLowerCase())
    .filter(Boolean);
  if (mentionTokens.some((token) => content.includes(token))) return true;

  if (!selfWxid) return false;
  const atWxids = extractAtWxids(m.raw);
  return atWxids.includes(selfWxid.toLowerCase());
}

function extractAtWxids(raw: unknown): string[] {
  if (!raw || typeof raw !== "object") return [];
  const o = raw as Record<string, unknown>;
  const direct = o.atWxids ?? o.ats ?? o.atList ?? o.at;
  if (Array.isArray(direct)) {
    return direct
      .map((v) => String(v).trim().toLowerCase())
      .filter(Boolean);
  }
  if (typeof direct === "string") {
    return direct
      .split(",")
      .map((s) => s.trim().toLowerCase())
      .filter(Boolean);
  }
  return [];
}

// ----------------------------- helpers ----------------------------

function makeLogger(runtime: RuntimeEnv | undefined): Logger {
  const log = (prefix: string) =>
    (...args: unknown[]) => {
      const line = `[${CHANNEL_ID}${prefix}] ${args.map(stringify).join(" ")}`;
      runtime?.log?.(line);
    };
  const err = (...args: unknown[]) => {
    const line = `[${CHANNEL_ID}] ${args.map(stringify).join(" ")}`;
    (runtime?.error ?? runtime?.log)?.(line);
  };
  return {
    info: log(""),
    warn: log(":warn"),
    error: err,
    debug: log(":debug"),
  };
}

function stringify(x: unknown): string {
  if (typeof x === "string") return x;
  if (x instanceof Error) return x.message;
  try {
    return JSON.stringify(x);
  } catch {
    return String(x);
  }
}

function noopResult(): MonitorResult {
  return { shutdown: async () => {} };
}