import assert from 'node:assert/strict';
import crypto from 'node:crypto';
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 studentUserId = '00000000-0000-0000-0000-000000000101';
const planId = '00000000-0000-0000-0000-000000000201';
const regionId = '00000000-0000-0000-0000-000000000301';

const ids = {
  paymentOrder: '10000000-0000-0000-0000-000000000701',
  payment: '10000000-0000-0000-0000-000000000702',
  refundOrder: '10000000-0000-0000-0000-000000000703',
  refundPayment: '10000000-0000-0000-0000-000000000704',
  refundRequest: '10000000-0000-0000-0000-000000000705',
  refundEntitlement: '10000000-0000-0000-0000-000000000706',
  providerBillJobWechat: '10000000-0000-0000-0000-000000000707',
  providerBillJobAlipay: '10000000-0000-0000-0000-000000000708',
  providerBillOrderWechat: '10000000-0000-0000-0000-000000000709',
  providerBillPaymentWechat: '10000000-0000-0000-0000-000000000710',
  providerBillOrderAlipay: '10000000-0000-0000-0000-000000000711',
  providerBillPaymentAlipay: '10000000-0000-0000-0000-000000000712',
};

const orderNos = {
  payment: 'CW-WX-PAY-001',
  refund: 'CW-WX-REFUND-001',
  providerBillWechat: 'CW-WX-BILL-001',
  providerBillAlipay: 'CW-ALI-BILL-001',
};
const refundNo = 'RF-CW-WX-001';
const providerBillDate = '2026-06-29';

const paymentFixture = (() => {
  const wechatMerchant = crypto.generateKeyPairSync('rsa', { modulusLength: 2048 });
  const alipayApp = crypto.generateKeyPairSync('rsa', { modulusLength: 2048 });
  return {
    wechatMerchantPrivateKey: wechatMerchant.privateKey.export({ type: 'pkcs8', format: 'pem' }).toString(),
    alipayAppPrivateKey: alipayApp.privateKey.export({ type: 'pkcs8', format: 'pem' }).toString(),
    wechatApiV3Key: '12345678901234567890123456789012',
  };
})();

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 startFakeWechatPayServer() {
  const port = await getFreePort();
  const baseUrl = `http://127.0.0.1:${port}`;
  const requests = [];
  const billRows = [
    {
      transactionType: 'payment',
      orderNo: orderNos.providerBillWechat,
      providerTradeNo: `wx-bill-trade-${orderNos.providerBillWechat}`,
      amountCents: 990,
      providerStatus: 'SUCCESS',
      paidAt: '2026-06-29T10:00:00+08:00',
    },
    {
      transactionType: 'payment',
      orderNo: 'CW-WX-PROVIDER-ONLY',
      providerTradeNo: 'wx-provider-only-trade',
      amountCents: 1888,
      providerStatus: 'SUCCESS',
    },
  ];
  const billBody = Buffer.from(JSON.stringify({ rows: billRows }), 'utf8');
  const billHash = crypto.createHash('sha256').update(billBody).digest('hex');
  const server = http.createServer((req, res) => {
    const url = new URL(req.url || '/', baseUrl);
    let raw = '';
    req.on('data', chunk => {
      raw += chunk.toString();
    });
    req.on('end', () => {
      requests.push({ method: req.method, pathname: url.pathname, search: url.search, body: raw ? JSON.parse(raw) : {} });

      if (req.method === 'GET' && url.pathname.startsWith('/v3/pay/transactions/out-trade-no/')) {
        const outTradeNo = decodeURIComponent(url.pathname.split('/').pop() || '');
        res.writeHead(200, { 'content-type': 'application/json' });
        res.end(JSON.stringify({
          appid: 'wx-worker-appid',
          mchid: 'wx-worker-mchid',
          out_trade_no: outTradeNo,
          transaction_id: `wx-trade-${outTradeNo}`,
          trade_state: 'SUCCESS',
          success_time: '2026-06-29T08:00:00+08:00',
          amount: { total: outTradeNo === orderNos.refund ? 500 : 990, currency: 'CNY' },
        }));
        return;
      }

      if (req.method === 'GET' && url.pathname.startsWith('/v3/refund/domestic/refunds/')) {
        const outRefundNo = decodeURIComponent(url.pathname.split('/').pop() || '');
        res.writeHead(200, { 'content-type': 'application/json' });
        res.end(JSON.stringify({
          refund_id: `refund-${outRefundNo}`,
          out_refund_no: outRefundNo,
          status: 'SUCCESS',
          amount: { refund: 500, total: 500, currency: 'CNY' },
          success_time: '2026-06-29T09:00:00+08:00',
        }));
        return;
      }

      if (req.method === 'GET' && url.pathname === '/v3/bill/tradebill') {
        res.writeHead(200, { 'content-type': 'application/json' });
        res.end(JSON.stringify({
          hash_type: 'SHA256',
          hash_value: billHash,
          download_url: `${baseUrl}/download/wechat-tradebill.json`,
        }));
        return;
      }

      if (req.method === 'GET' && url.pathname === '/download/wechat-tradebill.json') {
        res.writeHead(200, { 'content-type': 'application/json' });
        res.end(billBody);
        return;
      }

      res.writeHead(404, { 'content-type': 'application/json' });
      res.end(JSON.stringify({ code: 'NOT_FOUND' }));
    });
  });
  await new Promise((resolve, reject) => {
    server.once('error', reject);
    server.listen(port, '127.0.0.1', resolve);
  });
  return {
    paymentQueryEndpoint: `${baseUrl}/v3/pay/transactions/out-trade-no`,
    refundEndpoint: `${baseUrl}/v3/refund/domestic/refunds`,
    tradeBillEndpoint: `${baseUrl}/v3/bill/tradebill`,
    requests,
    close: () => new Promise(resolve => server.close(resolve)),
  };
}

async function startFakeAlipayServer() {
  const port = await getFreePort();
  const baseUrl = `http://127.0.0.1:${port}`;
  const requests = [];
  const billRows = [
    {
      transactionType: 'payment',
      orderNo: orderNos.providerBillAlipay,
      providerTradeNo: `ali-bill-trade-${orderNos.providerBillAlipay}`,
      amountCents: 1288,
      providerStatus: 'TRADE_SUCCESS',
      paidAt: '2026-06-29 11:00:00',
    },
  ];
  const billBody = Buffer.from(JSON.stringify({ rows: billRows }), 'utf8');
  const server = http.createServer((req, res) => {
    const url = new URL(req.url || '/', baseUrl);
    let raw = '';
    req.on('data', chunk => {
      raw += chunk.toString();
    });
    req.on('end', () => {
      requests.push({ method: req.method, pathname: url.pathname, search: url.search, body: raw });
      if (req.method === 'POST' && url.pathname === '/gateway.do') {
        res.writeHead(200, { 'content-type': 'application/json' });
        res.end(JSON.stringify({
          alipay_data_dataservice_bill_downloadurl_query_response: {
            code: '10000',
            msg: 'Success',
            bill_download_url: `${baseUrl}/download/alipay-tradebill.json`,
          },
        }));
        return;
      }
      if (req.method === 'GET' && url.pathname === '/download/alipay-tradebill.json') {
        res.writeHead(200, { 'content-type': 'application/json' });
        res.end(billBody);
        return;
      }
      res.writeHead(404, { 'content-type': 'application/json' });
      res.end(JSON.stringify({ code: 'NOT_FOUND' }));
    });
  });
  await new Promise((resolve, reject) => {
    server.once('error', reject);
    server.listen(port, '127.0.0.1', resolve);
  });
  return {
    endpoint: `${baseUrl}/gateway.do`,
    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', 'commerce'], {
    cwd: process.cwd(),
    env: {
      ...process.env,
      DATABASE_URL: databaseUrl,
      WORKER_COMMERCE_BATCH_SIZE: '20',
      WORKER_COMMERCE_MIN_AGE_SECONDS: '0',
      WORKER_COMMERCE_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, /commerce batch processed=\d+/, 'worker output should include commerce summary');
  assert.ok(!output.includes(paymentFixture.wechatApiV3Key), 'worker output must not leak WeChat API v3 key');
  assert.ok(!output.includes('PRIVATE KEY'), 'worker output must not leak merchant private key');
  return output;
}

async function runProviderBillsWorkerOnce() {
  const child = spawn(process.execPath, ['apps/worker/dist/apps/worker/src/index.js', '--once', '--job', 'provider-bills'], {
    cwd: process.cwd(),
    env: {
      ...process.env,
      DATABASE_URL: databaseUrl,
      WORKER_PROVIDER_BILL_BATCH_SIZE: '10',
      WORKER_PROVIDER_BILL_ID: 'provider-bills-integration-test',
      WORKER_COMMERCE_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, `provider bill worker should exit 0\n${output}`);
  assert.match(output, /provider-bills batch processed=\d+/, 'worker output should include provider bill summary');
  assert.ok(!output.includes(paymentFixture.wechatApiV3Key), 'provider bill worker output must not leak WeChat API v3 key');
  assert.ok(!output.includes('PRIVATE KEY'), 'provider bill worker output must not leak private keys');
  return output;
}

async function cleanup(pool) {
  await pool.query(
    `
    delete from public.commerce_bill_download_jobs
    where tenant_id = $1
      and (
        id in ($2::uuid, $3::uuid)
        or source_name like 'provider-bill:%:2026-06-29:%'
      )
    `,
    [tenantId, ids.providerBillJobWechat, ids.providerBillJobAlipay],
  );
  await pool.query(
    `
    delete from public.commerce_reconciliation_batches
    where tenant_id = $1
      and source = 'provider_download'
      and source_name like 'provider-bill:%:2026-06-29:%'
    `,
    [tenantId],
  );
  await pool.query(
    `
    delete from public.payment_events
    where tenant_id = $1
      and (
        event_id like $2
        or event_id like $3
      )
    `,
    [tenantId, `${tenantId}:payment-query:CW-WX-%`, `${tenantId}:refund-query:RF-CW-%`],
  );
  await pool.query(
    `
    delete from public.commerce_refund_events
    where tenant_id = $1 and refund_request_id = $2
    `,
    [tenantId, ids.refundRequest],
  );
  await pool.query(
    `
    delete from public.commerce_refund_requests
    where tenant_id = $1 and id = $2
    `,
    [tenantId, ids.refundRequest],
  );
  await pool.query(
    `
    delete from public.entitlements
    where tenant_id = $1
      and (
        id = $2
        or source_id in ($3::uuid, $4::uuid)
      )
    `,
    [tenantId, ids.refundEntitlement, ids.paymentOrder, ids.refundOrder],
  );
  await pool.query(
    `
    delete from public.payments
    where tenant_id = $1 and id in ($2::uuid, $3::uuid, $4::uuid, $5::uuid)
    `,
    [tenantId, ids.payment, ids.refundPayment, ids.providerBillPaymentWechat, ids.providerBillPaymentAlipay],
  );
  await pool.query(
    `
    delete from public.orders
    where tenant_id = $1 and id in ($2::uuid, $3::uuid, $4::uuid, $5::uuid)
    `,
    [tenantId, ids.paymentOrder, ids.refundOrder, ids.providerBillOrderWechat, ids.providerBillOrderAlipay],
  );
}

async function seed(pool, fakeWechat, fakeAlipay) {
  await pool.query(
    `
    insert into app_private.tenant_secrets (
      tenant_id, secret_scope, secret_key, secret_json, provider, last_rotated_at
    )
    values ($1, 'payment', 'wechat_pay', $2::jsonb, 'wechat_pay', now())
    on conflict (tenant_id, secret_scope, secret_key)
    do update set secret_json = excluded.secret_json,
                  provider = excluded.provider,
                  last_rotated_at = now(),
                  updated_at = now()
    `,
    [tenantId, JSON.stringify({
      privateKey: paymentFixture.wechatMerchantPrivateKey,
      apiV3Key: paymentFixture.wechatApiV3Key,
    })],
  );

  await pool.query(
    `
    insert into app_private.tenant_secrets (
      tenant_id, secret_scope, secret_key, secret_json, provider, last_rotated_at
    )
    values ($1, 'payment', 'alipay', $2::jsonb, 'alipay', now())
    on conflict (tenant_id, secret_scope, secret_key)
    do update set secret_json = excluded.secret_json,
                  provider = excluded.provider,
                  last_rotated_at = now(),
                  updated_at = now()
    `,
    [tenantId, JSON.stringify({
      privateKey: paymentFixture.alipayAppPrivateKey,
    })],
  );

  await pool.query(
    `
    insert into public.tenant_payment_accounts (
      tenant_id, provider, mode, display_name, status, config_public
    )
    values ($1, 'wechat_pay', 'tenant_collect', 'Worker WeChat Pay', 'active', $2::jsonb)
    on conflict (tenant_id, provider)
    do update set mode = excluded.mode,
                  display_name = excluded.display_name,
                  status = excluded.status,
                  config_public = excluded.config_public,
                  updated_at = now()
    `,
    [tenantId, JSON.stringify({
      appId: 'wx-worker-appid',
      merchantId: 'wx-worker-mchid',
      merchantSerialNo: 'wx-worker-serial',
      paymentQueryEndpoint: fakeWechat.paymentQueryEndpoint,
      refundEndpoint: fakeWechat.refundEndpoint,
      refundQueryEndpoint: fakeWechat.refundEndpoint,
      tradeBillEndpoint: fakeWechat.tradeBillEndpoint,
      secretRef: 'app_private.tenant_secrets:payment:wechat_pay',
    })],
  );

  await pool.query(
    `
    insert into public.tenant_payment_accounts (
      tenant_id, provider, mode, display_name, status, config_public
    )
    values ($1, 'alipay', 'tenant_collect', 'Worker Alipay', 'active', $2::jsonb)
    on conflict (tenant_id, provider)
    do update set mode = excluded.mode,
                  display_name = excluded.display_name,
                  status = excluded.status,
                  config_public = excluded.config_public,
                  updated_at = now()
    `,
    [tenantId, JSON.stringify({
      appId: 'ali-worker-appid',
      endpoint: fakeAlipay.endpoint,
      billEndpoint: fakeAlipay.endpoint,
      secretRef: 'app_private.tenant_secrets:payment:alipay',
    })],
  );

  await pool.query(
    `
    insert into public.orders (
      id, tenant_id, user_id, order_no, status, product_type, product_name,
      amount_cents, pay_method, pay_provider, plan_id, days, region_id, raw_payload,
      created_at, updated_at
    )
    values
      ($1, $2, $3, $4, 'pending', 'svip', 'Worker 补偿月卡', 990, 'jsapi', 'wechat_pay', $5, 30, $6, '{"source":"commerce-worker-test"}'::jsonb, now() - interval '10 minutes', now() - interval '10 minutes'),
      ($7, $2, $3, $8, 'paid', 'svip', 'Worker 退款月卡', 500, 'jsapi', 'wechat_pay', $5, 30, $6, '{"source":"commerce-worker-test"}'::jsonb, now() - interval '10 minutes', now() - interval '10 minutes'),
      ($9, $2, $3, $10, 'paid', 'svip', 'Worker 微信账单月卡', 990, 'jsapi', 'wechat_pay', $5, 30, $6, '{"source":"provider-bill-worker-test"}'::jsonb, now() - interval '10 minutes', now() - interval '10 minutes'),
      ($11, $2, $3, $12, 'paid', 'svip', 'Worker 支付宝账单月卡', 1288, 'wap', 'alipay', $5, 30, $6, '{"source":"provider-bill-worker-test"}'::jsonb, now() - interval '10 minutes', now() - interval '10 minutes')
    `,
    [
      ids.paymentOrder,
      tenantId,
      studentUserId,
      orderNos.payment,
      planId,
      regionId,
      ids.refundOrder,
      orderNos.refund,
      ids.providerBillOrderWechat,
      orderNos.providerBillWechat,
      ids.providerBillOrderAlipay,
      orderNos.providerBillAlipay,
    ],
  );

  await pool.query(
    `
    insert into public.payments (
      id, tenant_id, order_id, provider, method, status, amount_cents,
      provider_trade_no, paid_at, raw_payload, created_at, updated_at
    )
    values
      ($1, $2, $3, 'wechat_pay', 'jsapi', 'pending', 990, null, null, '{"source":"commerce-worker-test"}'::jsonb, now() - interval '10 minutes', now() - interval '10 minutes'),
      ($4, $2, $5, 'wechat_pay', 'jsapi', 'paid', 500, $6, now() - interval '9 minutes', '{"source":"commerce-worker-test"}'::jsonb, now() - interval '10 minutes', now() - interval '10 minutes'),
      ($7, $2, $8, 'wechat_pay', 'jsapi', 'paid', 990, $9, now() - interval '9 minutes', '{"source":"provider-bill-worker-test"}'::jsonb, now() - interval '10 minutes', now() - interval '10 minutes'),
      ($10, $2, $11, 'alipay', 'wap', 'paid', 1288, $12, now() - interval '9 minutes', '{"source":"provider-bill-worker-test"}'::jsonb, now() - interval '10 minutes', now() - interval '10 minutes')
    `,
    [
      ids.payment,
      tenantId,
      ids.paymentOrder,
      ids.refundPayment,
      ids.refundOrder,
      `wx-trade-${orderNos.refund}`,
      ids.providerBillPaymentWechat,
      ids.providerBillOrderWechat,
      `wx-bill-trade-${orderNos.providerBillWechat}`,
      ids.providerBillPaymentAlipay,
      ids.providerBillOrderAlipay,
      `ali-bill-trade-${orderNos.providerBillAlipay}`,
    ],
  );

  await pool.query(
    `
    insert into public.entitlements (
      id, tenant_id, user_id, entitlement_type, scope_type, scope_id,
      source_type, source_id, starts_at, expires_at, status, metadata
    )
    values ($1, $2, $3, 'svip', 'region', $4, 'order', $5, now() - interval '9 minutes', now() + interval '30 days', 'active', '{"source":"commerce-worker-test"}'::jsonb)
    `,
    [ids.refundEntitlement, tenantId, studentUserId, regionId, ids.refundOrder],
  );

  await pool.query(
    `
    insert into public.commerce_refund_requests (
      id, tenant_id, order_id, payment_id, refund_no, provider, provider_refund_no,
      status, amount_cents, reason, entitlement_action,
      requested_by, reviewed_by, processed_by,
      requested_at, reviewed_at, processed_at, metadata
    )
    values (
      $1, $2, $3, $4, $5, 'wechat_pay', null,
      'processing', 500, 'commerce worker integration refund', 'revoke_on_success',
      null, null, null,
      now() - interval '8 minutes', now() - interval '8 minutes', now() - interval '8 minutes',
      '{"source":"commerce-worker-test"}'::jsonb
    )
    `,
    [ids.refundRequest, tenantId, ids.refundOrder, ids.refundPayment, refundNo],
  );

  await pool.query(
    `
    insert into public.commerce_bill_download_jobs (
      id, tenant_id, provider, bill_date, bill_type, status,
      source_name, requested_by, metadata
    )
    values
      ($1, $3, 'wechat_pay', $4::date, 'payment', 'queued', $5, null, '{"source":"commerce-worker-test"}'::jsonb),
      ($2, $3, 'alipay', $4::date, 'payment', 'queued', $6, null, '{"source":"commerce-worker-test"}'::jsonb)
    on conflict (tenant_id, provider, bill_date, bill_type)
    where status in ('queued', 'running', 'completed')
    do update set status = 'queued',
                  source_name = excluded.source_name,
                  reconciliation_batch_id = null,
                  source_hash = null,
                  row_count = 0,
                  claimed_by = null,
                  claimed_at = null,
                  completed_at = null,
                  failed_at = null,
                  error_code = null,
                  error_message = null,
                  metadata = excluded.metadata,
                  updated_at = now()
    `,
    [
      ids.providerBillJobWechat,
      ids.providerBillJobAlipay,
      tenantId,
      providerBillDate,
      `provider-bill:wechat_pay:${providerBillDate}:payment`,
      `provider-bill:alipay:${providerBillDate}:payment`,
    ],
  );
}

async function main() {
  const fakeWechat = await startFakeWechatPayServer();
  const fakeAlipay = await startFakeAlipayServer();
  const pool = new pg.Pool({ connectionString: databaseUrl });
  let seeded = false;
  try {
    await pool.query('begin');
    await cleanup(pool);
    await seed(pool, fakeWechat, fakeAlipay);
    await pool.query('commit');
    seeded = true;

    const output = await runWorkerOnce();
    assert.match(output, /paid=1/, 'worker should reconcile one paid payment');
    assert.match(output, /succeeded=1/, 'worker should reconcile one succeeded refund');

    assert.ok(
      fakeWechat.requests.some(item => item.method === 'GET' && item.pathname.endsWith(`/${orderNos.payment}`)),
      'worker should query WeChat payment by out_trade_no',
    );
    assert.ok(
      fakeWechat.requests.some(item => item.method === 'GET' && item.pathname.endsWith(`/${refundNo}`)),
      'worker should query WeChat refund by out_refund_no',
    );

    const paymentOrder = await pool.query(
      `
      select o.status, o.trade_no, p.status as payment_status, p.provider_trade_no,
             count(e.id)::int as entitlement_count
      from public.orders o
      join public.payments p on p.tenant_id = o.tenant_id and p.order_id = o.id
      left join public.entitlements e on e.tenant_id = o.tenant_id and e.source_type = 'order' and e.source_id = o.id
      where o.tenant_id = $1 and o.id = $2
      group by o.status, o.trade_no, p.status, p.provider_trade_no
      `,
      [tenantId, ids.paymentOrder],
    );
    assert.equal(paymentOrder.rows[0]?.status, 'paid', 'payment compensation should mark order paid');
    assert.equal(paymentOrder.rows[0]?.payment_status, 'paid', 'payment compensation should mark payment paid');
    assert.equal(paymentOrder.rows[0]?.trade_no, `wx-trade-${orderNos.payment}`, 'payment compensation should record trade number');
    assert.equal(paymentOrder.rows[0]?.entitlement_count, 1, 'payment compensation should grant one order entitlement');

    const refund = await pool.query(
      `
      select rr.status, rr.provider_refund_no, o.status as order_status,
             o.refunded_amount_cents, p.status as payment_status,
             p.refunded_amount_cents as payment_refunded_amount_cents,
             e.status as entitlement_status
      from public.commerce_refund_requests rr
      join public.orders o on o.tenant_id = rr.tenant_id and o.id = rr.order_id
      join public.payments p on p.tenant_id = rr.tenant_id and p.id = rr.payment_id
      left join public.entitlements e on e.tenant_id = rr.tenant_id and e.source_type = 'order' and e.source_id = rr.order_id
      where rr.tenant_id = $1 and rr.id = $2
      `,
      [tenantId, ids.refundRequest],
    );
    assert.equal(refund.rows[0]?.status, 'succeeded', 'refund compensation should mark refund succeeded');
    assert.equal(refund.rows[0]?.provider_refund_no, `refund-${refundNo}`, 'refund compensation should record provider refund number');
    assert.equal(refund.rows[0]?.order_status, 'refunded', 'full refund compensation should mark order refunded');
    assert.equal(refund.rows[0]?.payment_status, 'refunded', 'full refund compensation should mark payment refunded');
    assert.equal(refund.rows[0]?.refunded_amount_cents, 500, 'full refund compensation should update order refunded amount');
    assert.equal(refund.rows[0]?.payment_refunded_amount_cents, 500, 'full refund compensation should update payment refunded amount');
    assert.equal(refund.rows[0]?.entitlement_status, 'revoked', 'full refund compensation should revoke order entitlement');

    await runWorkerOnce();
    const secondPass = await pool.query(
      `
      select count(*)::int as entitlement_count
      from public.entitlements
      where tenant_id = $1 and source_type = 'order' and source_id = $2
      `,
      [tenantId, ids.paymentOrder],
    );
    assert.equal(secondPass.rows[0]?.entitlement_count, 1, 'second worker pass must not grant duplicate entitlement');

    const events = await pool.query(
      `
      select payload, error
      from public.payment_events
      where tenant_id = $1
        and (
          event_id = $2
          or event_id = $3
        )
      order by created_at desc
      `,
      [
        tenantId,
        `${tenantId}:payment-query:${orderNos.payment}`,
        `${tenantId}:refund-query:${refundNo}`,
      ],
    );
    assert.ok(events.rows.length >= 2, 'worker should record payment events for reconciliation');
    assert.ok(!JSON.stringify(events.rows).includes(paymentFixture.wechatApiV3Key), 'payment event payload must not leak API v3 key');
    assert.ok(!JSON.stringify(events.rows).includes('PRIVATE KEY'), 'payment event payload must not leak private key');

    const providerBillOutput = await runProviderBillsWorkerOnce();
    assert.match(providerBillOutput, /completed=2/, 'provider bill worker should complete two provider bill jobs');
    assert.ok(
      fakeWechat.requests.some(item => item.method === 'GET' && item.pathname === '/v3/bill/tradebill'),
      'provider bill worker should request WeChat trade bill URL',
    );
    assert.ok(
      fakeWechat.requests.some(item => item.method === 'GET' && item.pathname === '/download/wechat-tradebill.json'),
      'provider bill worker should download WeChat trade bill',
    );
    assert.ok(
      fakeAlipay.requests.some(item => item.method === 'POST' && item.pathname === '/gateway.do' && item.body.includes('alipay.data.dataservice.bill.downloadurl.query')),
      'provider bill worker should request Alipay bill download URL',
    );
    assert.ok(
      fakeAlipay.requests.some(item => item.method === 'GET' && item.pathname === '/download/alipay-tradebill.json'),
      'provider bill worker should download Alipay trade bill',
    );

    const providerJobs = await pool.query(
      `
      select provider, status, source_hash, row_count, download_url_host,
             reconciliation_batch_id, error_code, error_message
      from public.commerce_bill_download_jobs
      where tenant_id = $1 and id in ($2::uuid, $3::uuid)
      order by provider
      `,
      [tenantId, ids.providerBillJobWechat, ids.providerBillJobAlipay],
    );
    assert.equal(providerJobs.rows.length, 2, 'provider bill jobs should be persisted');
    assert.ok(providerJobs.rows.every(row => row.status === 'completed'), 'provider bill jobs should be completed');
    assert.ok(providerJobs.rows.every(row => row.source_hash && !row.error_code && !row.error_message), 'completed provider bill jobs should keep source hash and no errors');
    assert.ok(providerJobs.rows.every(row => row.download_url_host === '127.0.0.1'), 'provider bill jobs should persist only download URL host');

    const providerBatches = await pool.query(
      `
      select b.id, b.provider, b.source, b.source_name, b.status,
             b.matched_count, b.missing_local_count, b.total_count,
             count(i.id)::int as item_count
      from public.commerce_reconciliation_batches b
      join public.commerce_reconciliation_items i
        on i.tenant_id = b.tenant_id and i.batch_id = b.id
      where b.tenant_id = $1
        and b.source = 'provider_download'
        and b.source_name like 'provider-bill:%:2026-06-29:%'
      group by b.id
      order by b.provider
      `,
      [tenantId],
    );
    assert.equal(providerBatches.rows.length, 2, 'provider bill downloads should create two reconciliation batches');
    assert.ok(
      providerBatches.rows.some(row => row.provider === 'wechat_pay' && row.status === 'completed_with_issues' && row.matched_count >= 1 && row.missing_local_count >= 1),
      'WeChat bill batch should include matched and provider-only anomaly rows',
    );
    assert.ok(
      providerBatches.rows.some(row => row.provider === 'alipay' && ['completed', 'completed_with_issues'].includes(row.status) && row.matched_count >= 1),
      'Alipay bill batch should include matched rows',
    );

    const jobPayload = JSON.stringify(providerJobs.rows);
    assert.ok(!jobPayload.includes(paymentFixture.wechatApiV3Key), 'provider bill job rows must not leak API v3 key');
    assert.ok(!jobPayload.includes('PRIVATE KEY'), 'provider bill job rows must not leak private keys');

    console.log('Commerce worker integration test complete.');
  } catch (error) {
    await pool.query('rollback').catch(() => {});
    throw error;
  } finally {
    if (seeded) {
      await cleanup(pool).catch(() => {});
    }
    await pool.end();
    await fakeWechat.close();
    await fakeAlipay.close();
  }
}

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