gongxue-base/scripts/crm-worker-integration-test.js

import assert from 'node:assert/strict';
import http from 'node:http';
import pg from 'pg';
import { spawn } from 'node:child_process';

const databaseUrl = process.env.DATABASE_URL || 'postgresql://postgres:postgres@127.0.0.1:54322/postgres';
const tenantId = '00000000-0000-0000-0000-000000000001';
const adminUserId = '00000000-0000-0000-0000-000000000102';
const studentUserId = '00000000-0000-0000-0000-000000000101';
const salesUserId = '00000000-0000-0000-0000-000000000104';

function getFreePort() {
  return new Promise((resolve, reject) => {
    const server = http.createServer();
    server.listen(0, '127.0.0.1', () => {
      const address = server.address();
      server.close(() => resolve(address.port));
    });
    server.on('error', reject);
  });
}

async function startWebhookServer() {
  const port = await getFreePort();
  const requests = [];
  const server = http.createServer((req, res) => {
    let raw = '';
    req.on('data', chunk => {
      raw += chunk.toString();
    });
    req.on('end', () => {
      requests.push({
        url: req.url,
        headers: req.headers,
        body: raw ? JSON.parse(raw) : {},
      });
      res.writeHead(200, { 'content-type': 'application/json' });
      res.end(JSON.stringify({ ok: true }));
    });
  });
  await new Promise(resolve => server.listen(port, '127.0.0.1', resolve));
  return {
    url: `http://127.0.0.1:${port}/crm`,
    requests,
    close: () => new Promise(resolve => server.close(resolve)),
  };
}

async function runWorkerOnce() {
  const child = spawn(process.execPath, ['apps/worker/dist/apps/worker/src/index.js', '--once', '--job', 'crm'], {
    cwd: process.cwd(),
    env: {
      ...process.env,
      DATABASE_URL: databaseUrl,
      WORKER_CRM_ALLOW_INSECURE_LOCALHOST: 'true',
      WORKER_CRM_REQUEST_TIMEOUT_MS: '5000',
    },
    stdio: ['ignore', 'pipe', 'pipe'],
    windowsHide: true,
  });
  let output = '';
  child.stdout.on('data', chunk => {
    output += chunk.toString();
  });
  child.stderr.on('data', chunk => {
    output += chunk.toString();
  });
  const code = await new Promise(resolve => child.on('exit', resolve));
  assert.equal(code, 0, `worker should exit 0\n${output}`);
  assert.match(output, /processed=\d+/, 'worker output should include processed count');
}

async function main() {
  const webhook = await startWebhookServer();
  const pool = new pg.Pool({ connectionString: databaseUrl });
  try {
    await pool.query('begin');
    await pool.query(
      `
      delete from public.crm_webhook_log
      where tenant_id = $1 and record_id = $2
      `,
      [tenantId, studentUserId],
    );
    await pool.query(
      `
      delete from public.crm_webhook_queue
      where tenant_id = $1 and idempotency_key = 'lead:worker-integration'
      `,
      [tenantId],
    );
    await pool.query(
      `
      insert into app_private.tenant_secrets (tenant_id, secret_scope, secret_key, secret_value, provider, last_rotated_at)
      values ($1, 'crm', 'webhook', 'crm-secret-worker-test', 'webhook', now())
      on conflict (tenant_id, secret_scope, secret_key)
      do update set secret_value = excluded.secret_value,
                    provider = excluded.provider,
                    last_rotated_at = now(),
                    updated_at = now()
      `,
      [tenantId],
    );
    await pool.query(
      `
      insert into public.crm_config (
        tenant_id, enabled, url, secret_ref, form_name, exam_type, timeout_sec, delay_sec
      )
      values ($1, true, $2, 'app_private.tenant_secrets:crm:webhook', 'Worker集成测试客资', '专升本', 5, 0)
      on conflict (tenant_id)
      do update set enabled = excluded.enabled,
                    url = excluded.url,
                    secret_ref = excluded.secret_ref,
                    form_name = excluded.form_name,
                    exam_type = excluded.exam_type,
                    timeout_sec = excluded.timeout_sec,
                    delay_sec = excluded.delay_sec,
                    updated_at = now()
      `,
      [tenantId, webhook.url],
    );
    await pool.query(
      `
      insert into public.referral_leads (
        tenant_id, student_user_id, referrer_user_id, ref_code, source, bind_type, status, metadata
      )
      values ($1, $2, $3, 'WORKER1'::citext, 'integration', 'manual', 'protected', '{"source":"crm-worker-test"}'::jsonb)
      on conflict (tenant_id, student_user_id)
      do update set referrer_user_id = excluded.referrer_user_id,
                    ref_code = excluded.ref_code,
                    source = excluded.source,
                    bind_type = excluded.bind_type,
                    status = excluded.status,
                    updated_at = now()
      `,
      [tenantId, studentUserId, salesUserId],
    );
    const lead = await pool.query(
      `
      select id, bound_at
      from public.referral_leads
      where tenant_id = $1 and student_user_id = $2
      limit 1
      `,
      [tenantId, studentUserId],
    );
    await pool.query(
      `
      insert into public.crm_webhook_queue (
        tenant_id, record_id, status, scheduled_at, lead_id, source,
        payload, idempotency_key, target_url
      )
      values (
        $1, $2, 'pending', now() - interval '1 second', $3, 'integration',
        $4::jsonb, 'lead:worker-integration', $5
      )
      on conflict (tenant_id, idempotency_key) where idempotency_key is not null
      do update set status = 'pending',
                    scheduled_at = excluded.scheduled_at,
                    next_attempt_at = null,
                    attempts = 0,
                    last_error = null,
                    payload = excluded.payload,
                    target_url = excluded.target_url,
                    updated_at = now()
      `,
      [
        tenantId,
        studentUserId,
        lead.rows[0].id,
        JSON.stringify({
          provider: 'generic',
          formName: 'Worker集成测试客资',
          examType: '专升本',
          source: 'integration',
          leadId: lead.rows[0].id,
          student: { id: studentUserId, name: '烟测学生', phone: '13800000000' },
          referral: { refCode: 'WORKER1', referrerUserId: salesUserId, boundAt: lead.rows[0].bound_at },
        }),
        webhook.url,
      ],
    );
    await pool.query('commit');

    await runWorkerOnce();
    assert.equal(webhook.requests.length, 1, 'worker should send exactly one webhook request');
    assert.equal(webhook.requests[0].body.event, 'lead.created', 'generic webhook should send lead.created event');
    assert.equal(webhook.requests[0].body.data.student.phone, '13800000000', 'webhook body should include lead phone');
    assert.ok(!JSON.stringify(webhook.requests[0]).includes('crm-secret-worker-test'), 'webhook request should not leak CRM secret for generic provider');

    const queue = await pool.query(
      `
      select status, attempts, last_http_code, sent_at
      from public.crm_webhook_queue
      where tenant_id = $1 and idempotency_key = 'lead:worker-integration'
      limit 1
      `,
      [tenantId],
    );
    assert.equal(queue.rows[0]?.status, 'sent', 'CRM queue item should be sent');
    assert.equal(queue.rows[0]?.attempts, 1, 'CRM queue item should record one attempt');
    assert.equal(queue.rows[0]?.last_http_code, 200, 'CRM queue item should record HTTP 200');
    assert.ok(queue.rows[0]?.sent_at, 'CRM queue item should record sent_at');

    const logs = await pool.query(
      `
      select outcome, http_code, request_payload, response_summary
      from public.crm_webhook_log
      where tenant_id = $1 and record_id = $2
      order by created_at desc
      limit 1
      `,
      [tenantId, studentUserId],
    );
    assert.equal(logs.rows[0]?.outcome, 'sent', 'CRM log should record sent outcome');
    assert.equal(logs.rows[0]?.http_code, 200, 'CRM log should record HTTP 200');
    assert.equal(logs.rows[0]?.request_payload?.provider, 'generic', 'CRM log should record provider');
    assert.ok(!JSON.stringify(logs.rows[0]).includes('crm-secret-worker-test'), 'CRM log should not leak secret');

    console.log('CRM worker integration test complete.');
  } finally {
    await pool.end();
    await webhook.close();
  }
}

main().catch(error => {
  console.error(error);
  process.exit(1);
});