openclaw/public-release/vk-gateway/vk-longpoll.mjs

#!/usr/bin/env node
import fs from 'node:fs';
import { setTimeout as sleep } from 'node:timers/promises';

function parseArgs() {
  const args = process.argv.slice(2);
  const out = { env: null };
  for (let i = 0; i < args.length; i++) {
    if (args[i] === '--env') out.env = args[++i];
  }
  return out;
}

function loadEnv(path) {
  if (!path) return;
  const txt = fs.readFileSync(path, 'utf8');
  for (const line of txt.split(/\r?\n/)) {
    const t = line.trim();
    if (!t || t.startsWith('#')) continue;
    const idx = t.indexOf('=');
    if (idx === -1) continue;
    const k = t.slice(0, idx).trim();
    const v = t.slice(idx + 1).trim();
    if (!(k in process.env)) process.env[k] = v;
  }
}

function parseCsv(value, fallback = []) {
  if (!value) return fallback;
  return value
    .split(',')
    .map((x) => x.trim())
    .filter(Boolean);
}

const { env } = parseArgs();
loadEnv(env);

const CFG = {
  groupId: Number(process.env.VK_GROUP_ID || 0),
  token: process.env.VK_TOKEN || '',
  apiVersion: process.env.VK_API_VERSION || '5.199',
  wait: Number(process.env.VK_WAIT || 25),
  bridgeUrl: process.env.OPENCLAW_BRIDGE_URL || '',
  bridgeTimeout: Number(process.env.OPENCLAW_BRIDGE_TIMEOUT_MS || 45000),
  allowedEvents: new Set(parseCsv(process.env.VK_ALLOWED_EVENTS, ['message_new'])),
  syncSettings: process.env.VK_SYNC_LONGPOLL_SETTINGS === '1',
};

if (!CFG.groupId || !CFG.token || !CFG.bridgeUrl) {
  console.error('Missing required env: VK_GROUP_ID, VK_TOKEN, OPENCLAW_BRIDGE_URL');
  process.exit(1);
}
if (CFG.apiVersion !== '5.199') {
  console.warn(`[vk-bridge] warning: VK_API_VERSION=${CFG.apiVersion}; recommended 5.199`);
}
if (CFG.wait > 90 || CFG.wait < 1) {
  console.error('VK_WAIT must be between 1 and 90');
  process.exit(1);
}

const seen = new Set();

function vkApi(method, params = {}) {
  const qs = new URLSearchParams({
    ...Object.fromEntries(Object.entries(params).map(([k, v]) => [k, String(v)])),
    access_token: CFG.token,
    v: CFG.apiVersion,
  });
  return fetch(`https://api.vk.com/method/${method}?${qs}`).then((r) => r.json());
}

async function getLongPollServer() {
  const data = await vkApi('groups.getLongPollServer', { group_id: CFG.groupId });
  if (data.error) throw new Error(`groups.getLongPollServer: ${JSON.stringify(data.error)}`);
  return data.response;
}

async function getLongPollSettings() {
  const data = await vkApi('groups.getLongPollSettings', { group_id: CFG.groupId });
  if (data.error) throw new Error(`groups.getLongPollSettings: ${JSON.stringify(data.error)}`);
  return data.response;
}

async function setLongPollSettings(params) {
  const data = await vkApi('groups.setLongPollSettings', {
    group_id: CFG.groupId,
    ...params,
  });
  if (data.error) throw new Error(`groups.setLongPollSettings: ${JSON.stringify(data.error)}`);
  return data.response;
}

async function ensureLongPollSettings() {
  const settings = await getLongPollSettings();
  const updates = { enabled: 1, api_version: CFG.apiVersion };

  for (const eventType of CFG.allowedEvents) {
    const current = settings?.events?.[eventType];
    if (current !== 1) updates[eventType] = 1;
  }

  const needApply = Object.keys(updates).length > 2 || settings?.is_enabled !== 1;
  if (!needApply) {
    console.log('[vk-bridge] long poll settings already OK');
    return;
  }

  await setLongPollSettings(updates);
  console.log('[vk-bridge] long poll settings synced');
}

async function callBridge(evt) {
  const msg = evt?.object?.message;
  if (!msg) return { silent: true };

  const body = {
    source: 'vk',
    group_id: CFG.groupId,
    event_type: evt?.type || null,
    event_id: evt?.event_id || null,
    user_id: msg.from_id,
    peer_id: msg.peer_id,
    text: msg.text || '',
    payload: msg.payload || null,
    attachments: msg.attachments || [],
    raw: evt,
  };

  const ctrl = new AbortController();
  const t = setTimeout(() => ctrl.abort(), CFG.bridgeTimeout);
  try {
    const res = await fetch(CFG.bridgeUrl, {
      method: 'POST',
      headers: { 'content-type': 'application/json' },
      body: JSON.stringify(body),
      signal: ctrl.signal,
    });
    if (!res.ok) throw new Error(`Bridge HTTP ${res.status}`);
    return await res.json();
  } finally {
    clearTimeout(t);
  }
}

async function sendVk(peerId, text) {
  const randomId = Math.floor(Math.random() * 2_147_483_647);
  const data = await vkApi('messages.send', {
    peer_id: peerId,
    random_id: randomId,
    message: text,
  });
  if (data.error) throw new Error(`messages.send: ${JSON.stringify(data.error)}`);
  return data.response;
}

function isInboundUserMessage(evt) {
  if (!CFG.allowedEvents.has(evt?.type)) return false;
  if (evt?.type !== 'message_new') return false;
  const msg = evt?.object?.message;
  if (!msg) return false;
  if ((msg.out ?? 0) === 1) return false;
  if (!msg.from_id || !msg.peer_id) return false;
  return true;
}

async function main() {
  if (CFG.syncSettings) {
    try {
      await ensureLongPollSettings();
    } catch (e) {
      console.error('[vk-bridge] failed to sync long poll settings:', e.message);
    }
  }

  let lp = await getLongPollServer();
  let ts = lp.ts;

  console.log(`[vk-bridge] started: group=${CFG.groupId}`);

  while (true) {
    try {
      const qs = new URLSearchParams({
        act: 'a_check',
        key: lp.key,
        ts: String(ts),
        wait: String(CFG.wait),
      });
      const r = await fetch(`${lp.server}?${qs}`);
      const data = await r.json();

      if (data.failed) {
        if (data.failed === 1 && data.ts) {
          ts = data.ts;
          continue;
        }
        lp = await getLongPollServer();
        ts = lp.ts;
        continue;
      }

      ts = data.ts;
      for (const evt of data.updates || []) {
        if (!isInboundUserMessage(evt)) continue;

        const msg = evt.object.message;
        const dedupKey = `${msg.peer_id}:${msg.conversation_message_id || msg.id}`;
        if (seen.has(dedupKey)) continue;
        seen.add(dedupKey);
        if (seen.size > 5000) {
          const first = seen.values().next().value;
          if (first) seen.delete(first);
        }

        try {
          const bridged = await callBridge(evt);
          if (!bridged || bridged.silent) continue;
          const reply = String(bridged.reply || '').trim();
          if (!reply) continue;
          await sendVk(msg.peer_id, reply);
        } catch (err) {
          console.error('[vk-bridge] handler error:', err.message);
        }
      }
    } catch (err) {
      console.error('[vk-bridge] loop error:', err.message);
      await sleep(1500);
      try {
        lp = await getLongPollServer();
        ts = lp.ts;
      } catch {
        await sleep(3000);
      }
    }
  }
}

main().catch((e) => {
  console.error('[vk-bridge] fatal:', e);
  process.exit(1);
});