QIWEI-skill/runtime/callback-service/src/personal-polling.mjs

import path from 'node:path';
import { spawn } from 'node:child_process';
import { PACKAGE_ROOT } from './config-loader.mjs';

let processorBridgePromise = null;

function loadProcessorBridge() {
  if (!processorBridgePromise) processorBridgePromise = import('./processor-bridge.mjs');
  return processorBridgePromise;
}

const defaultListenerApi = {
  getStatus: async () => (await loadProcessorBridge()).getPersonalListenerStatus(),
  recoverLogin: async () => (await loadProcessorBridge()).recoverPersonalLogin(),
  start: async () => (await loadProcessorBridge()).startPersonalListener(),
  stop: async options => (await loadProcessorBridge()).stopPersonalListener(options),
};

export class PersonalPollingRuntime {
  constructor({ config, workspaceRoot, onState = () => {}, listenerApi = {} }) {
    this.config = config;
    this.workspaceRoot = workspaceRoot;
    this.onState = onState;
    this.listenerApi = {
      getStatus: listenerApi.getStatus || defaultListenerApi.getStatus,
      recoverLogin: listenerApi.recoverLogin || defaultListenerApi.recoverLogin,
      start: listenerApi.start || defaultListenerApi.start,
      stop: listenerApi.stop || defaultListenerApi.stop,
    };
    this.running = false;
    this.listenerStarted = false;
    this.friendWorker = null;
    this.loopPromise = null;
    this.cancelWait = null;
    this.lastRecoveryAttemptAt = 0;
    this.lastRecoveryError = '';
  }

  wait(ms) {
    return new Promise(resolve => {
      const timer = setTimeout(() => {
        this.cancelWait = null;
        resolve();
      }, ms);
      this.cancelWait = () => {
        clearTimeout(timer);
        this.cancelWait = null;
        resolve();
      };
    });
  }

  startFriendWorker() {
    if (!this.config.friendPollingEnabled || this.friendWorker) return;
    const scriptPath = path.join(PACKAGE_ROOT, 'scripts', 'friend-polling-worker.js');
    this.friendWorker = spawn(process.execPath, [scriptPath], {
      cwd: this.workspaceRoot,
      env: { ...process.env, QIWEI_WORKSPACE_ROOT: this.workspaceRoot },
      stdio: 'inherit',
      windowsHide: true,
    });
    this.onState({ friendPolling: { status: 'running', pid: this.friendWorker.pid } });
    this.friendWorker.once('exit', (code, signal) => {
      this.friendWorker = null;
      this.onState({ friendPolling: { status: this.running ? 'error' : 'stopped', code, signal } });
    });
  }

  recoveryDue() {
    const cooldownMs = Math.max(30000, Number(this.config.retryMs) || 10000);
    return Date.now() - this.lastRecoveryAttemptAt >= cooldownMs;
  }

  async recoverAndRestart(listener = {}) {
    if (!this.recoveryDue()) return null;
    this.lastRecoveryAttemptAt = Date.now();
    this.onState({ personalPolling: { status: 'reconnecting' } });
    const recovered = await this.listenerApi.recoverLogin();
    const recoveredCode = Number(recovered?.summary?.statusCode);
    if (!recovered?.summary?.loggedIn && recoveredCode !== 2) {
      this.lastRecoveryError = String(
        recovered?.errors?.[0]?.message
        || recovered?.assistantMessage
        || 'Automatic login recovery is pending.',
      );
      return null;
    }

    this.lastRecoveryError = '';
    if (listener.running) await this.listenerApi.stop({ preserveAgentState: true });
    this.listenerStarted = false;
    const started = await this.listenerApi.start();
    this.listenerStarted = Boolean(started?.data?.running);
    return started;
  }

  async pollOnce() {
    let result = await this.listenerApi.getStatus();
    let listener = result?.data?.listener || {};
    let account = result?.data?.account || {};
    let requiresLogin = account.reason === 'upstream_device_missing';

    if (account.online === false && !requiresLogin) {
      await this.recoverAndRestart(listener);
      result = await this.listenerApi.getStatus();
      listener = result?.data?.listener || {};
      account = result?.data?.account || {};
      requiresLogin = account.reason === 'upstream_device_missing';
    }

    if (!listener.running && account.online !== false) {
      try {
        const started = await this.listenerApi.start();
        this.listenerStarted = Boolean(started?.data?.running);
      } catch (error) {
        const recovered = await this.recoverAndRestart(listener);
        if (!recovered) throw error;
      }
      result = await this.listenerApi.getStatus();
      listener = result?.data?.listener || {};
      account = result?.data?.account || {};
    }

    this.listenerStarted = Boolean(listener.running);
    if (this.listenerStarted) this.startFriendWorker();
    const listenerHealthy = listener.running && account.online !== false;
    if (listenerHealthy) this.lastRecoveryError = '';
    this.onState({
      personalPolling: {
        status: listenerHealthy ? 'running' : (requiresLogin ? 'needs_login' : 'waiting'),
        syncKey: Number(listener.syncKey || 0),
        startedAt: listener.startedAt || null,
        lastError: listenerHealthy
          ? ''
          : (requiresLogin ? account.statusText : '') || listener.lastError || this.lastRecoveryError,
      },
    });
    return listener;
  }

  async loop() {
    while (this.running) {
      try {
        await this.pollOnce();
      } catch (error) {
        this.listenerStarted = false;
        this.onState({ personalPolling: { status: 'waiting', lastError: error.message } });
      }
      if (this.running) await this.wait(this.config.retryMs);
    }
  }

  start() {
    if (this.running) return;
    this.running = true;
    this.onState({ personalPolling: { status: 'starting', lastError: '' } });
    this.loopPromise = this.loop();
  }

  async stop() {
    this.running = false;
    if (this.cancelWait) this.cancelWait();
    if (this.listenerStarted) {
      try { await this.listenerApi.stop({ preserveAgentState: true }); } catch {}
    }
    this.listenerStarted = false;
    if (this.friendWorker) {
      try { this.friendWorker.kill(); } catch {}
      this.friendWorker = null;
    }
    this.onState({
      personalPolling: { status: 'stopped' },
      friendPolling: { status: 'stopped' },
    });
    await this.loopPromise;
  }
}