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';

const PORT = 4100;
const BASE = `http://127.0.0.1:${PORT}`;
const ADMIN_KEY = 'relay-smoke-admin';
const CALLBACK_SECRET = 'relay-smoke-callback';
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,
  },
  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 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);

  process.stdout.write(JSON.stringify({
    status: 'ok',
    globalRouting: true,
    largePayloadBytes: 5000,
    duplicateProtected: true,
    missingGuidRejected: true,
    invalidSignatureRejected: true,
  }, null, 2));
} finally {
  child.kill('SIGTERM');
  fs.rmSync(tempRoot, { recursive: true, force: true });
}