openclaw-voc-skill/services/qiwei-relay/scripts/global-ingest-smoke-test.mjs

import assert from 'node:assert/strict';
import crypto from 'node:crypto';
import fs from 'node:fs';
import os from 'node:os';
import path from 'node:path';
import { spawn } from 'node:child_process';
import { createServer } from 'node:http';

const PORT = 4100;
const BASE = `http://127.0.0.1:${PORT}`;
const ADMIN_KEY = 'relay-smoke-admin';
const CALLBACK_SECRET = 'relay-smoke-callback';
const FMODE_PORT = 4101;
const fmodeServer = createServer((req, res) => {
  const authorization = String(req.headers.authorization || '');
  const userId = authorization.includes('account-b') ? 'fmode-account-b' : 'fmode-account-a';
  res.writeHead(200, { 'Content-Type': 'application/json' });
  res.end(JSON.stringify({ data: { userId } }));
});
await new Promise((resolve, reject) => {
  fmodeServer.once('error', reject);
  fmodeServer.listen(FMODE_PORT, '127.0.0.1', resolve);
});
const tempRoot = fs.mkdtempSync(path.join(os.tmpdir(), 'qiwei-relay-smoke-'));
const child = spawn(process.execPath, ['dist/server.js'], {
  cwd: path.resolve(import.meta.dirname, '..'),
  env: {
    ...process.env,
    PORT: String(PORT),
    DB_PATH: path.join(tempRoot, 'relay.db'),
    ADMIN_KEY,
    UPSTREAM_CALLBACK_SECRET: CALLBACK_SECRET,
    RELAY_PUBLIC_URL: BASE,
    FMODE_GATEWAY_URL: `http://127.0.0.1:${FMODE_PORT}`,
  },
  stdio: ['ignore', 'pipe', 'pipe'],
});

async function request(route, { method = 'GET', body, headers = {} } = {}) {
  const response = await fetch(`${BASE}${route}`, {
    method,
    headers: {
      ...(body === undefined ? {} : { 'Content-Type': 'application/json' }),
      ...headers,
    },
    body: body === undefined ? undefined : JSON.stringify(body),
  });
  return { response, data: await response.json().catch(() => ({})) };
}

async function waitForHealth() {
  const deadline = Date.now() + 15000;
  while (Date.now() < deadline) {
    try {
      const { response } = await request('/api/health');
      if (response.ok) return;
    } catch {}
    await new Promise(resolve => setTimeout(resolve, 150));
  }
  throw new Error('Relay smoke server did not become healthy');
}

function decrypt(encryptedPayload, privateKeyPem) {
  assert.ok(encryptedPayload.startsWith('v2:'));
  const envelope = JSON.parse(Buffer.from(encryptedPayload.slice(3), 'base64').toString('utf8'));
  const privateKey = crypto.createPrivateKey(privateKeyPem);
  const aesKey = crypto.privateDecrypt(
    { key: privateKey, oaepHash: 'sha256' },
    Buffer.from(envelope.key, 'base64'),
  );
  const decipher = crypto.createDecipheriv('aes-256-gcm', aesKey, Buffer.from(envelope.iv, 'base64'));
  decipher.setAuthTag(Buffer.from(envelope.tag, 'base64'));
  return Buffer.concat([
    decipher.update(Buffer.from(envelope.ciphertext, 'base64')),
    decipher.final(),
  ]).toString('utf8');
}

try {
  await waitForHealth();
  const tenantResponse = await request('/api/admin/tenant', {
    method: 'POST',
    headers: { 'X-Admin-Key': ADMIN_KEY },
    body: { name: 'global-ingest-smoke' },
  });
  assert.equal(tenantResponse.response.status, 200);
  const tenant = tenantResponse.data;
  const guid = `smoke-guid-${crypto.randomUUID()}`;
  const deviceResponse = await request('/api/tenant/device', {
    method: 'POST',
    headers: { Authorization: `Bearer ${tenant.apiSecret}` },
    body: { guid, deviceName: 'smoke-device' },
  });
  assert.equal(deviceResponse.response.status, 200);

  const callbackBody = {
    code: 0,
    data: [{
      guid,
      msgUniqueIdentifier: `smoke-${crypto.randomUUID()}`,
      msgType: 1,
      content: 'x'.repeat(5000),
    }],
  };
  const rawBody = JSON.stringify(callbackBody);
  const signature = crypto.createHmac('sha256', CALLBACK_SECRET).update(rawBody).digest('hex');
  const ingest = await request('/api/webhook/ingest', {
    method: 'POST',
    headers: { 'X-Signature': signature },
    body: callbackBody,
  });
  assert.equal(ingest.response.status, 200);
  assert.equal(ingest.data.queued, 1);

  const duplicate = await request('/api/webhook/ingest', {
    method: 'POST',
    headers: { 'X-Signature': signature },
    body: callbackBody,
  });
  assert.equal(duplicate.response.status, 200);
  assert.equal(duplicate.data.duplicates, 1);

  const rawAuthorization = await request('/api/webhook/ingest', {
    method: 'POST',
    headers: { Authorization: CALLBACK_SECRET },
    body: callbackBody,
  });
  assert.equal(rawAuthorization.response.status, 200);
  assert.equal(rawAuthorization.data.duplicates, 1);

  const setupProbe = await request('/api/webhook/ingest', {
    method: 'POST',
    headers: { Authorization: CALLBACK_SECRET },
    body: { msg: '设置订阅成功!!', callBackUrl: 'http://relay.example/api/webhook/ingest' },
  });
  assert.equal(setupProbe.response.status, 200);
  assert.equal(setupProbe.data.probe, true);
  assert.equal(setupProbe.data.queued, 0);

  const poll = await request('/api/relay/poll', {
    method: 'POST',
    headers: { Authorization: `Bearer ${tenant.apiSecret}` },
    body: { guid, batchSize: 10, waitMs: 1000 },
  });
  assert.equal(poll.response.status, 200);
  assert.equal(poll.data.events.length, 1);
  const decrypted = JSON.parse(decrypt(poll.data.events[0].encryptedPayload, tenant.privateKey));
  assert.equal(decrypted.data[0].guid, guid);
  assert.equal(decrypted.data[0].content.length, 5000);

  const ack = await request('/api/relay/ack', {
    method: 'POST',
    headers: { Authorization: `Bearer ${tenant.apiSecret}` },
    body: { guid, eventIds: [poll.data.events[0].eventId] },
  });
  assert.equal(ack.response.status, 200);
  assert.equal(ack.data.ackedCount, 1);

  const missingGuid = await request('/api/webhook/ingest', {
    method: 'POST',
    headers: {
      'X-Signature': crypto.createHmac('sha256', CALLBACK_SECRET).update('{}').digest('hex'),
    },
    body: {},
  });
  assert.equal(missingGuid.response.status, 400);

  const invalidSignature = await request('/api/webhook/ingest', {
    method: 'POST',
    headers: { 'X-Signature': 'invalid' },
    body: callbackBody,
  });
  assert.equal(invalidSignature.response.status, 401);

  const transferGuid = `transfer-guid-${crypto.randomUUID()}`;
  const accountTenantResponse = await request('/api/tenant/register', {
    method: 'POST',
    headers: { Authorization: 'Bearer account-a-token' },
    body: { description: 'account-a-owner', deviceGuid: transferGuid },
  });
  assert.equal(accountTenantResponse.response.status, 200);

  const replacementTenantResponse = await request('/api/admin/tenant', {
    method: 'POST',
    headers: { 'X-Admin-Key': ADMIN_KEY },
    body: { name: 'account-a-replacement' },
  });
  assert.equal(replacementTenantResponse.response.status, 200);
  const replacementTenant = replacementTenantResponse.data;
  const transfer = await request('/api/tenant/device', {
    method: 'POST',
    headers: {
      Authorization: `Bearer ${replacementTenant.apiSecret}`,
      'X-Fmode-Token': 'account-a-token',
    },
    body: { guid: transferGuid },
  });
  assert.equal(transfer.response.status, 200);

  const oldTenantStatus = await request('/api/tenant/status', {
    headers: { Authorization: `Bearer ${accountTenantResponse.data.apiSecret}` },
  });
  const replacementStatus = await request('/api/tenant/status', {
    headers: { Authorization: `Bearer ${replacementTenant.apiSecret}` },
  });
  assert.equal(oldTenantStatus.data.devices.some(item => item.guid === transferGuid), false);
  assert.equal(replacementStatus.data.devices.some(item => item.guid === transferGuid), true);

  const foreignTenantResponse = await request('/api/admin/tenant', {
    method: 'POST',
    headers: { 'X-Admin-Key': ADMIN_KEY },
    body: { name: 'account-b' },
  });
  const crossAccountTransfer = await request('/api/tenant/device', {
    method: 'POST',
    headers: {
      Authorization: `Bearer ${foreignTenantResponse.data.apiSecret}`,
      'X-Fmode-Token': 'account-b-token',
    },
    body: { guid: transferGuid },
  });
  assert.equal(crossAccountTransfer.response.status, 409);

  const secondAccountAGuid = `account-a-second-${crypto.randomUUID()}`;
  const accountBGuid = `account-b-device-${crypto.randomUUID()}`;
  const secondAccountADevice = await request('/api/tenant/device', {
    method: 'POST',
    headers: {
      Authorization: `Bearer ${replacementTenant.apiSecret}`,
      'X-Fmode-Token': 'account-a-token',
    },
    body: { guid: secondAccountAGuid },
  });
  const accountBDevice = await request('/api/tenant/device', {
    method: 'POST',
    headers: {
      Authorization: `Bearer ${foreignTenantResponse.data.apiSecret}`,
      'X-Fmode-Token': 'account-b-token',
    },
    body: { guid: accountBGuid },
  });
  assert.equal(secondAccountADevice.response.status, 200);
  assert.equal(accountBDevice.response.status, 200);

  const multiDeviceBody = {
    code: 0,
    data: [
      { guid: transferGuid, cmd: 15000, msgType: 2, msgUniqueIdentifier: `multi-${crypto.randomUUID()}`, msgData: { content: 'account-a-primary' } },
      { guid: secondAccountAGuid, cmd: 15000, msgType: 2, msgUniqueIdentifier: `multi-${crypto.randomUUID()}`, msgData: { content: 'account-a-secondary' } },
      { guid: accountBGuid, cmd: 15000, msgType: 2, msgUniqueIdentifier: `multi-${crypto.randomUUID()}`, msgData: { content: 'account-b-primary' } },
    ],
  };
  const multiDeviceRaw = JSON.stringify(multiDeviceBody);
  const multiDeviceIngest = await request('/api/webhook/ingest', {
    method: 'POST',
    headers: { 'X-Signature': crypto.createHmac('sha256', CALLBACK_SECRET).update(multiDeviceRaw).digest('hex') },
    body: multiDeviceBody,
  });
  assert.equal(multiDeviceIngest.response.status, 200);
  assert.equal(multiDeviceIngest.data.queued, 3);

  for (const [deviceGuid, apiSecret, privateKey, expectedContent] of [
    [transferGuid, replacementTenant.apiSecret, replacementTenant.privateKey, 'account-a-primary'],
    [secondAccountAGuid, replacementTenant.apiSecret, replacementTenant.privateKey, 'account-a-secondary'],
    [accountBGuid, foreignTenantResponse.data.apiSecret, foreignTenantResponse.data.privateKey, 'account-b-primary'],
  ]) {
    const devicePoll = await request('/api/relay/poll', {
      method: 'POST',
      headers: { Authorization: `Bearer ${apiSecret}` },
      body: { guid: deviceGuid, batchSize: 10, waitMs: 1000 },
    });
    assert.equal(devicePoll.response.status, 200);
    assert.equal(devicePoll.data.events.length, 1);
    const devicePayload = JSON.parse(decrypt(devicePoll.data.events[0].encryptedPayload, privateKey));
    assert.equal(devicePayload.data[0].msgData.content, expectedContent);
  }

  process.stdout.write(JSON.stringify({
    status: 'ok',
    globalRouting: true,
    largePayloadBytes: 5000,
    duplicateProtected: true,
    missingGuidRejected: true,
    invalidSignatureRejected: true,
    sameAccountDeviceTransfer: true,
    crossAccountDeviceTransferRejected: true,
    multipleDevicesPerAccount: true,
    multipleAccountsIsolated: true,
  }, null, 2));
} finally {
  if (child.exitCode === null) {
    const exited = new Promise(resolve => child.once('exit', resolve));
    child.kill('SIGTERM');
    await Promise.race([exited, new Promise(resolve => setTimeout(resolve, 3000))]);
  }
  await new Promise(resolve => fmodeServer.close(resolve));
  fs.rmSync(tempRoot, { recursive: true, force: true, maxRetries: 5, retryDelay: 100 });
}