import crypto from 'node:crypto';
import JSZip from 'jszip';
import iconv from 'iconv-lite';
import type pg from 'pg';
import { pool } from '../db.js';
import { config } from '../config.js';
import {
  importReconciliationBatch,
  normalizeReconciliationBillDate,
  normalizeReconciliationBillType,
  normalizeReconciliationProvider,
  type BillType,
  type ReconciliationProvider,
} from '../../../api/src/features/commerce/reconciliation.js';

type ProviderBillProvider = Exclude<ReconciliationProvider, 'manual'>;

interface SecretRow {
  secretValue: string | null;
  secretJson: Record<string, unknown> | null;
}

interface ProviderConfig {
  tenantId: string;
  provider: ProviderBillProvider;
  configPublic: Record<string, unknown>;
  secret: SecretRow | null;
}

interface ProviderBillJob {
  id: string;
  tenantId: string;
  provider: ProviderBillProvider;
  billDate: string;
  billType: BillType;
  sourceName: string;
  requestedBy: string | null;
  metadata: Record<string, unknown>;
}

interface DownloadedBill {
  rows: Record<string, unknown>[];
  sourceHash: string;
  downloadHashType: string | null;
  downloadHashValue: string | null;
  downloadUrlHost: string | null;
  metadata: Record<string, unknown>;
}

interface ProviderBillWorkerResult {
  processed: number;
  completed: number;
  failed: number;
  skipped: number;
}

class ProviderBillWorkerError extends Error {
  readonly code: string;

  constructor(message: string, code = 'PROVIDER_BILL_WORKER_ERROR') {
    super(message);
    this.code = code;
  }
}

function objectValue(value: unknown): Record<string, unknown> {
  return value && typeof value === 'object' && !Array.isArray(value) ? value as Record<string, unknown> : {};
}

function safeString(value: unknown) {
  return typeof value === 'string' && value.trim() ? value.trim() : '';
}

function optionalPublicString(providerConfig: ProviderConfig, keys: string[]) {
  for (const key of keys) {
    const value = providerConfig.configPublic[key];
    if (typeof value === 'string' && value.trim()) return value.trim();
  }
  return '';
}

function requirePublicString(providerConfig: ProviderConfig, keys: string[]) {
  const value = optionalPublicString(providerConfig, keys);
  if (value) return value;
  throw new ProviderBillWorkerError(`${providerConfig.provider} public config is missing ${keys[0]}`, 'PAYMENT_PUBLIC_CONFIG_REQUIRED');
}

function secretString(providerConfig: ProviderConfig, keys: string[]) {
  const secret = providerConfig.secret;
  if (!secret) return '';
  if (typeof secret.secretValue === 'string' && secret.secretValue.trim()) return secret.secretValue.trim();
  const json = objectValue(secret.secretJson);
  for (const key of keys) {
    const value = json[key];
    if (typeof value === 'string' && value.trim()) return value.trim();
  }
  return '';
}

function requireSecretString(providerConfig: ProviderConfig, keys: string[]) {
  const value = secretString(providerConfig, keys);
  if (value) return value;
  throw new ProviderBillWorkerError(`${providerConfig.provider} secret is missing ${keys[0]}`, 'PAYMENT_SECRET_REQUIRED');
}

function normalizePem(value: string, label: 'PRIVATE KEY' | 'PUBLIC KEY') {
  if (value.includes('-----BEGIN')) return value;
  const wrapped = value.match(/.{1,64}/g)?.join('\n') || value;
  return `-----BEGIN ${label}-----\n${wrapped}\n-----END ${label}-----`;
}

function rsaSignSha256(privateKey: string, message: string) {
  return crypto.createSign('RSA-SHA256').update(message, 'utf8').sign(privateKey, 'base64');
}

function randomNonce(size = 16) {
  return crypto.randomBytes(size).toString('base64url');
}

function canonicalForm(params: Record<string, string>) {
  return Object.keys(params)
    .filter(key => params[key] !== undefined && params[key] !== null && params[key] !== '')
    .sort()
    .map(key => `${key}=${params[key]}`)
    .join('&');
}

function encodedForm(params: Record<string, string>) {
  return Object.keys(params)
    .filter(key => params[key] !== undefined && params[key] !== null && params[key] !== '')
    .sort()
    .map(key => `${encodeURIComponent(key)}=${encodeURIComponent(params[key])}`)
    .join('&');
}

function parseSecretRef(value: unknown) {
  if (typeof value !== 'string' || !value.trim()) return null;
  const parts = value.trim().split(':');
  if (parts.length !== 3 || parts[0] !== 'app_private.tenant_secrets') return null;
  if (parts[1] !== 'payment' || !parts[2]) return null;
  return { scope: parts[1], key: parts[2] };
}

function normalizeProvider(value: string | null | undefined): ProviderBillProvider | null {
  const normalized = normalizeReconciliationProvider(String(value || ''));
  return normalized === 'manual' ? null : normalized;
}

function providerAliases(provider: ProviderBillProvider) {
  return provider === 'wechat_pay'
    ? ['wechat_pay', 'wechat-pay', 'wechatpay', 'wxpay']
    : ['alipay', 'ali_pay'];
}

function isLocalDevHost(hostname: string) {
  return ['127.0.0.1', 'localhost', '::1'].includes(hostname);
}

function isAllowedHost(hostname: string, allowedHosts: string[]) {
  return allowedHosts.some(host => hostname === host || hostname.endsWith(`.${host}`));
}

function providerEndpointForKeys(
  providerConfig: ProviderConfig,
  keys: string[],
  fallback: string,
  allowedHosts: string[],
  code = 'PAYMENT_ENDPOINT_NOT_ALLOWED',
) {
  const raw = optionalPublicString(providerConfig, keys) || fallback;
  let endpoint: URL;
  try {
    endpoint = new URL(raw);
  } catch {
    throw new ProviderBillWorkerError(`${providerConfig.provider} endpoint is invalid`, 'PAYMENT_ENDPOINT_INVALID');
  }

  const localDev = process.env.NODE_ENV !== 'production' && isLocalDevHost(endpoint.hostname);
  if (!localDev && endpoint.protocol !== 'https:') {
    throw new ProviderBillWorkerError(`${providerConfig.provider} endpoint must use HTTPS`, code);
  }
  if (!localDev && !isAllowedHost(endpoint.hostname, allowedHosts)) {
    throw new ProviderBillWorkerError(`${providerConfig.provider} endpoint host is not allowed`, code);
  }
  endpoint.username = '';
  endpoint.password = '';
  return endpoint.toString();
}

function assertDownloadUrlAllowed(url: string, provider: ProviderBillProvider) {
  let parsed: URL;
  try {
    parsed = new URL(url);
  } catch {
    throw new ProviderBillWorkerError('Provider bill download URL is invalid', 'PROVIDER_BILL_DOWNLOAD_URL_INVALID');
  }
  const localDev = process.env.NODE_ENV !== 'production' && isLocalDevHost(parsed.hostname);
  if (localDev) return parsed;
  if (parsed.protocol !== 'https:') {
    throw new ProviderBillWorkerError('Provider bill download URL must use HTTPS', 'PROVIDER_BILL_DOWNLOAD_URL_NOT_ALLOWED');
  }
  const hosts = provider === 'wechat_pay'
    ? ['api.mch.weixin.qq.com', 'download.mch.weixin.qq.com']
    : ['openapi.alipay.com', 'dwbillcenter.alipay.com', 'download.alipay.com'];
  if (!isAllowedHost(parsed.hostname, hosts)) {
    throw new ProviderBillWorkerError('Provider bill download URL host is not allowed', 'PROVIDER_BILL_DOWNLOAD_URL_NOT_ALLOWED');
  }
  return parsed;
}

async function loadSecret(client: pg.PoolClient, tenantId: string, configPublic: Record<string, unknown>, provider: ProviderBillProvider) {
  const secretRef = parseSecretRef(configPublic.secretRef);
  const secretKey = secretRef?.key || provider;
  const result = await client.query<SecretRow>(
    `
    select secret_value as "secretValue", secret_json as "secretJson"
    from app_private.tenant_secrets
    where tenant_id = $1
      and secret_scope = 'payment'
      and secret_key = $2
    limit 1
    `,
    [tenantId, secretKey],
  );
  return result.rows[0] || null;
}

async function loadPaymentProviderConfig(client: pg.PoolClient, tenantId: string, provider: ProviderBillProvider): Promise<ProviderConfig> {
  const aliases = providerAliases(provider);
  const accountResult = await client.query<{
    provider: string;
    configPublic: Record<string, unknown> | null;
  }>(
    `
    select provider, config_public as "configPublic"
    from public.tenant_payment_accounts
    where tenant_id = $1
      and provider = any($2::text[])
      and status = 'active'
    order by array_position($2::text[], provider)
    limit 1
    `,
    [tenantId, aliases],
  );
  const account = accountResult.rows[0];
  if (!account) throw new ProviderBillWorkerError(`${provider} provider is not configured`, 'PAYMENT_PROVIDER_NOT_CONFIGURED');
  const normalized = normalizeProvider(account.provider);
  if (!normalized) throw new ProviderBillWorkerError(`${account.provider} provider is unsupported`, 'PAYMENT_PROVIDER_NOT_SUPPORTED');
  const configPublic = objectValue(account.configPublic);
  const secret = await loadSecret(client, tenantId, configPublic, normalized);
  return { tenantId, provider: normalized, configPublic, secret };
}

async function fetchJson(url: string, init: RequestInit, timeoutMs: number) {
  const controller = new AbortController();
  const timeout = setTimeout(() => controller.abort(), timeoutMs);
  try {
    const response = await fetch(url, { ...init, signal: controller.signal });
    const raw = objectValue(await response.json().catch(() => ({})));
    return { response, raw };
  } finally {
    clearTimeout(timeout);
  }
}

async function fetchBuffer(url: string, init: RequestInit, timeoutMs: number) {
  const controller = new AbortController();
  const timeout = setTimeout(() => controller.abort(), timeoutMs);
  try {
    const response = await fetch(url, { ...init, signal: controller.signal });
    const buffer = Buffer.from(await response.arrayBuffer());
    return { response, buffer };
  } finally {
    clearTimeout(timeout);
  }
}

function wechatPrivateKey(providerConfig: ProviderConfig) {
  return normalizePem(requireSecretString(providerConfig, ['privateKey', 'merchantPrivateKey']), 'PRIVATE KEY');
}

function signedWechatGetHeaders(providerConfig: ProviderConfig, url: URL) {
  const mchId = requirePublicString(providerConfig, ['merchantId', 'mchId']);
  const merchantSerialNo = requirePublicString(providerConfig, ['merchantSerialNo']);
  const timestamp = Math.floor(Date.now() / 1000).toString();
  const nonce = randomNonce();
  const message = ['GET', `${url.pathname}${url.search}`, timestamp, nonce, ''].join('\n') + '\n';
  const signature = rsaSignSha256(wechatPrivateKey(providerConfig), message);
  return {
    accept: 'application/json',
    authorization: `WECHATPAY2-SHA256-RSA2048 mchid="${mchId}",nonce_str="${nonce}",signature="${signature}",timestamp="${timestamp}",serial_no="${merchantSerialNo}"`,
  };
}

function wechatBillEndpoint(providerConfig: ProviderConfig, billType: BillType) {
  if (billType === 'refund') {
    return providerEndpointForKeys(
      providerConfig,
      ['wechatRefundBillEndpoint', 'refundBillEndpoint', 'tradeBillEndpoint'],
      'https://api.mch.weixin.qq.com/v3/bill/tradebill',
      ['api.mch.weixin.qq.com'],
    );
  }
  return providerEndpointForKeys(
    providerConfig,
    ['wechatTradeBillEndpoint', 'tradeBillEndpoint'],
    'https://api.mch.weixin.qq.com/v3/bill/tradebill',
    ['api.mch.weixin.qq.com'],
  );
}

function mapWechatBillType(billType: BillType) {
  if (billType === 'refund') return 'REFUND';
  return 'ALL';
}

async function downloadWechatBill(providerConfig: ProviderConfig, job: ProviderBillJob): Promise<DownloadedBill> {
  const endpoint = wechatBillEndpoint(providerConfig, job.billType);
  const url = new URL(endpoint);
  url.searchParams.set('bill_date', job.billDate);
  url.searchParams.set('bill_type', mapWechatBillType(job.billType));
  const { response, raw } = await fetchJson(url.toString(), {
    method: 'GET',
    headers: signedWechatGetHeaders(providerConfig, url),
  }, config.commerceRequestTimeoutMs);
  if (!response.ok) throw new ProviderBillWorkerError('WeChat Pay bill URL request failed', 'PROVIDER_BILL_URL_REQUEST_FAILED');

  const downloadUrl = safeString(raw.download_url);
  if (!downloadUrl) throw new ProviderBillWorkerError('WeChat Pay bill response missing download_url', 'PROVIDER_BILL_RESPONSE_INVALID');
  const hashType = safeString(raw.hash_type) || null;
  const hashValue = safeString(raw.hash_value) || null;
  const parsedDownloadUrl = assertDownloadUrlAllowed(downloadUrl, 'wechat_pay');
  const { response: downloadResponse, buffer } = await fetchBuffer(downloadUrl, {
    method: 'GET',
    headers: signedWechatGetHeaders(providerConfig, parsedDownloadUrl),
  }, config.commerceRequestTimeoutMs);
  if (!downloadResponse.ok) throw new ProviderBillWorkerError('WeChat Pay bill download failed', 'PROVIDER_BILL_DOWNLOAD_FAILED');
  verifyProviderHash(buffer, hashType, hashValue);
  const rows = await parseBillRows(buffer, downloadResponse.headers.get('content-type') || '', parsedDownloadUrl.pathname, 'wechat_pay');
  return {
    rows,
    sourceHash: crypto.createHash('sha256').update(buffer).digest('hex'),
    downloadHashType: hashType,
    downloadHashValue: hashValue,
    downloadUrlHost: parsedDownloadUrl.hostname,
    metadata: {
      providerResponse: {
        hashType,
        hasHashValue: Boolean(hashValue),
      },
      downloadedBytes: buffer.length,
    },
  };
}

function alipayBillType(billType: BillType) {
  if (billType === 'refund') return 'trade';
  return 'trade';
}

async function downloadAlipayBill(providerConfig: ProviderConfig, job: ProviderBillJob): Promise<DownloadedBill> {
  const appId = requirePublicString(providerConfig, ['appId']);
  const gateway = providerEndpointForKeys(
    providerConfig,
    ['billEndpoint', 'endpoint'],
    'https://openapi.alipay.com/gateway.do',
    ['openapi.alipay.com'],
  );
  const privateKey = normalizePem(requireSecretString(providerConfig, ['privateKey', 'appPrivateKey']), 'PRIVATE KEY');
  const params: Record<string, string> = {
    app_id: appId,
    method: 'alipay.data.dataservice.bill.downloadurl.query',
    charset: 'utf-8',
    sign_type: 'RSA2',
    timestamp: new Date().toISOString().replace('T', ' ').slice(0, 19),
    version: '1.0',
    biz_content: JSON.stringify({
      bill_type: alipayBillType(job.billType),
      bill_date: job.billDate,
    }),
  };
  params.sign = rsaSignSha256(privateKey, canonicalForm(params));

  const { response, raw } = await fetchJson(gateway, {
    method: 'POST',
    headers: {
      accept: 'application/json',
      'content-type': 'application/x-www-form-urlencoded;charset=utf-8',
    },
    body: encodedForm(params),
  }, config.commerceRequestTimeoutMs);
  if (!response.ok) throw new ProviderBillWorkerError('Alipay bill URL request failed', 'PROVIDER_BILL_URL_REQUEST_FAILED');
  const responseBody = objectValue(raw.alipay_data_dataservice_bill_downloadurl_query_response);
  if (safeString(responseBody.code) !== '10000') {
    throw new ProviderBillWorkerError('Alipay bill URL request was rejected', 'PROVIDER_BILL_URL_REQUEST_REJECTED');
  }
  const downloadUrl = safeString(responseBody.bill_download_url);
  if (!downloadUrl) throw new ProviderBillWorkerError('Alipay bill response missing bill_download_url', 'PROVIDER_BILL_RESPONSE_INVALID');
  const parsedDownloadUrl = assertDownloadUrlAllowed(downloadUrl, 'alipay');
  const { response: downloadResponse, buffer } = await fetchBuffer(downloadUrl, {
    method: 'GET',
    headers: { accept: '*/*' },
  }, config.commerceRequestTimeoutMs);
  if (!downloadResponse.ok) throw new ProviderBillWorkerError('Alipay bill download failed', 'PROVIDER_BILL_DOWNLOAD_FAILED');
  const rows = await parseBillRows(buffer, downloadResponse.headers.get('content-type') || '', parsedDownloadUrl.pathname, 'alipay');
  return {
    rows,
    sourceHash: crypto.createHash('sha256').update(buffer).digest('hex'),
    downloadHashType: null,
    downloadHashValue: null,
    downloadUrlHost: parsedDownloadUrl.hostname,
    metadata: {
      providerResponse: {
        code: responseBody.code,
      },
      downloadedBytes: buffer.length,
    },
  };
}

function verifyProviderHash(buffer: Buffer, hashType: string | null, hashValue: string | null) {
  if (!hashType && !hashValue) return;
  if (!hashType || !hashValue) {
    throw new ProviderBillWorkerError('Provider bill hash metadata is incomplete', 'PROVIDER_BILL_HASH_INVALID');
  }
  const normalized = hashType.trim().toUpperCase();
  const algorithm = normalized === 'SHA1' ? 'sha1' : normalized === 'SHA256' ? 'sha256' : '';
  if (!algorithm) {
    throw new ProviderBillWorkerError(`Provider bill hash type ${hashType} is unsupported`, 'PROVIDER_BILL_HASH_UNSUPPORTED');
  }
  const actual = crypto.createHash(algorithm).update(buffer).digest('hex');
  if (actual.toLowerCase() !== hashValue.toLowerCase()) {
    throw new ProviderBillWorkerError('Provider bill hash mismatch', 'PROVIDER_BILL_HASH_MISMATCH');
  }
}

function decodeText(buffer: Buffer) {
  if (buffer.length >= 3 && buffer[0] === 0xef && buffer[1] === 0xbb && buffer[2] === 0xbf) {
    return buffer.subarray(3).toString('utf8');
  }
  const utf8 = buffer.toString('utf8');
  if (utf8.includes('�')) {
    return iconv.decode(buffer, 'gb18030');
  }
  return utf8;
}

async function parseBillRows(buffer: Buffer, contentType: string, pathname: string, provider: ProviderBillProvider) {
  const lowerPath = pathname.toLowerCase();
  const lowerType = contentType.toLowerCase();
  if (lowerPath.endsWith('.zip') || lowerType.includes('zip')) {
    const zip = await JSZip.loadAsync(buffer);
    const rows: Record<string, unknown>[] = [];
    for (const entry of Object.values(zip.files)) {
      if (entry.dir) continue;
      const name = entry.name.toLowerCase();
      if (!name.endsWith('.csv') && !name.endsWith('.txt') && !name.endsWith('.json')) continue;
      const entryBuffer = Buffer.from(await entry.async('uint8array'));
      rows.push(...parseBillRowsFromText(decodeText(entryBuffer), provider));
    }
    return rows;
  }
  return parseBillRowsFromText(decodeText(buffer), provider);
}

function parseBillRowsFromText(text: string, provider: ProviderBillProvider) {
  const trimmed = text.trim();
  if (!trimmed) return [];
  if (trimmed.startsWith('{') || trimmed.startsWith('[')) {
    const parsed = JSON.parse(trimmed) as unknown;
    const parsedObject = objectValue(parsed);
    const rows: unknown[] = Array.isArray(parsed)
      ? parsed
      : Array.isArray(parsedObject.rows)
        ? parsedObject.rows
        : [];
    return rows.map(row => objectValue(row));
  }
  return parseDelimitedBillRows(text, provider);
}

function parseDelimitedBillRows(text: string, provider: ProviderBillProvider) {
  const lines = text
    .split(/\r?\n/)
    .map(line => line.trim())
    .filter(line => line && !line.startsWith('#') && !line.startsWith('`----------'));
  const headerIndex = lines.findIndex(line => {
    const normalized = line.replace(/`/g, '');
    return normalized.includes(',') && (
      normalized.includes('商户订单号')
      || normalized.includes('out_trade_no')
      || normalized.includes('providerTradeNo')
      || normalized.includes('交易号')
      || normalized.includes('微信支付订单号')
    );
  });
  if (headerIndex < 0) {
    throw new ProviderBillWorkerError('Provider bill header not found', 'PROVIDER_BILL_PARSE_FAILED');
  }
  const delimiter = lines[headerIndex].includes('\t') ? '\t' : ',';
  const headers = splitDelimitedLine(lines[headerIndex], delimiter).map(cleanCell);
  const rows: Record<string, unknown>[] = [];
  for (const line of lines.slice(headerIndex + 1)) {
    const values = splitDelimitedLine(line, delimiter).map(cleanCell);
    if (values.length < 2) continue;
    const raw: Record<string, unknown> = {};
    headers.forEach((header, index) => {
      if (header) raw[header] = values[index] || '';
    });
    const normalized = provider === 'wechat_pay' ? normalizeWechatBillRow(raw) : normalizeAlipayBillRow(raw);
    if (Object.keys(normalized).length > 0) rows.push(normalized);
  }
  return rows;
}

function splitDelimitedLine(line: string, delimiter: string) {
  if (delimiter === '\t') return line.split('\t');
  const cells: string[] = [];
  let current = '';
  let quoted = false;
  for (let index = 0; index < line.length; index += 1) {
    const char = line[index];
    if (char === '"') {
      if (quoted && line[index + 1] === '"') {
        current += '"';
        index += 1;
      } else {
        quoted = !quoted;
      }
    } else if (char === ',' && !quoted) {
      cells.push(current);
      current = '';
    } else {
      current += char;
    }
  }
  cells.push(current);
  return cells;
}

function cleanCell(value: string) {
  return value.trim().replace(/^`/, '').replace(/^"|"$/g, '').trim();
}

function firstValue(row: Record<string, unknown>, keys: string[]) {
  for (const key of keys) {
    const value = row[key];
    if (typeof value === 'string' && value.trim()) return value.trim();
    if (typeof value === 'number' && Number.isFinite(value)) return String(value);
  }
  return '';
}

function normalizeAmount(value: string) {
  if (!value) return '';
  const cleaned = value.replace(/,/g, '').replace(/[￥¥]/g, '').trim();
  const parsed = Number(cleaned);
  return Number.isFinite(parsed) ? cleaned : '';
}

function rowTypeFromText(value: string) {
  const normalized = value.toLowerCase();
  if (normalized.includes('refund') || value.includes('退款')) return 'refund';
  return 'payment';
}

function normalizeWechatBillRow(row: Record<string, unknown>) {
  const transactionType = rowTypeFromText(firstValue(row, ['账单类型', '交易类型', 'transaction_type', 'type']));
  const amount = normalizeAmount(firstValue(row, ['订单金额(元)', '应结订单金额(元)', '总金额', 'amount', 'total_amount']));
  const refundAmount = normalizeAmount(firstValue(row, ['申请退款金额(元)', '退款金额(元)', 'refund_amount', 'refund_fee']));
  return {
    transactionType,
    providerTradeNo: firstValue(row, ['微信支付订单号', 'transaction_id', 'providerTradeNo']),
    providerRefundNo: firstValue(row, ['微信退款单号', 'refund_id', 'providerRefundNo']),
    orderNo: firstValue(row, ['商户订单号', 'out_trade_no', 'orderNo']),
    refundNo: firstValue(row, ['商户退款单号', 'out_refund_no', 'refundNo']),
    amountYuan: transactionType === 'payment' ? amount : '',
    refundAmount: transactionType === 'refund' ? (refundAmount || amount) : '',
    feeYuan: normalizeAmount(firstValue(row, ['手续费(元)', 'service_fee', 'fee'])),
    paidAt: firstValue(row, ['支付完成时间', '交易时间', 'success_time', 'paidAt']),
    refundedAt: firstValue(row, ['退款成功时间', 'refund_success_time', 'refundedAt']),
    providerStatus: firstValue(row, ['交易状态', '退款状态', 'trade_state', 'status']) || 'SUCCESS',
    rawProviderRow: row,
  };
}

function normalizeAlipayBillRow(row: Record<string, unknown>) {
  const transactionType = rowTypeFromText(firstValue(row, ['业务类型', '交易类型', 'transaction_type', 'type']));
  const amount = normalizeAmount(firstValue(row, ['订单金额（元）', '订单金额(元)', '收入（+元）', '收入(+元)', 'amount', 'total_amount']));
  const refundAmount = normalizeAmount(firstValue(row, ['退款金额（元）', '退款金额(元)', '支出（-元）', '支出(-元)', 'refund_amount']));
  return {
    transactionType,
    providerTradeNo: firstValue(row, ['支付宝交易号', '交易号', 'trade_no', 'providerTradeNo']),
    providerRefundNo: firstValue(row, ['支付宝退款单号', 'refund_id', 'providerRefundNo']),
    orderNo: firstValue(row, ['商户订单号', 'out_trade_no', 'orderNo']),
    refundNo: firstValue(row, ['商户退款单号', 'out_request_no', 'refundNo']),
    amountYuan: transactionType === 'payment' ? amount : '',
    refundAmount: transactionType === 'refund' ? (refundAmount || amount) : '',
    feeYuan: normalizeAmount(firstValue(row, ['服务费（元）', '服务费(元)', 'fee', 'service_fee'])),
    paidAt: firstValue(row, ['交易创建时间', '付款时间', 'gmt_payment', 'paidAt']),
    refundedAt: firstValue(row, ['退款时间', 'gmt_refund', 'refundedAt']),
    providerStatus: firstValue(row, ['交易状态', 'trade_status', 'status']) || 'TRADE_SUCCESS',
    rawProviderRow: row,
  };
}

async function claimProviderBillJobs(client: pg.PoolClient, limit = config.providerBillBatchSize) {
  const result = await client.query<ProviderBillJob>(
    `
    update public.commerce_bill_download_jobs j
    set status = 'running',
        claimed_by = $2,
        claimed_at = now(),
        updated_at = now(),
        error_code = null,
        error_message = null
    where j.id in (
      select id
      from public.commerce_bill_download_jobs
      where status = 'queued'
         or (
           status = 'running'
           and claimed_at < now() - ($3::int * interval '1 second')
         )
      order by created_at asc
      limit $1
      for update skip locked
    )
    returning id, tenant_id as "tenantId", provider,
              bill_date::text as "billDate", bill_type as "billType",
              source_name as "sourceName", requested_by as "requestedBy",
              metadata
    `,
    [limit, config.providerBillWorkerId, config.providerBillClaimStaleSeconds],
  );
  return result.rows.map(row => ({
    ...row,
    provider: normalizeProvider(row.provider) || 'wechat_pay',
    billDate: normalizeReconciliationBillDate(row.billDate),
    billType: normalizeReconciliationBillType(row.billType),
    metadata: objectValue(row.metadata),
  }));
}

async function processProviderBillJob(job: ProviderBillJob) {
  const client = await pool.connect();
  try {
    const providerConfig = await loadPaymentProviderConfig(client, job.tenantId, job.provider);
    const downloaded = job.provider === 'wechat_pay'
      ? await downloadWechatBill(providerConfig, job)
      : await downloadAlipayBill(providerConfig, job);
    await client.query('begin');
    const imported = await importReconciliationBatch(client, {
      tenantId: job.tenantId,
      actorUserId: job.requestedBy,
      provider: job.provider,
      billDate: job.billDate,
      billType: job.billType,
      source: 'provider_download',
      sourceName: `${job.sourceName}:${job.id}`,
      rows: downloaded.rows,
      metadata: {
        ...job.metadata,
        providerBillJobId: job.id,
        providerBillSourceHash: downloaded.sourceHash,
        providerBill: downloaded.metadata,
      },
    });
    await client.query(
      `
      update public.commerce_bill_download_jobs
      set status = 'completed',
          source_hash = $3,
          row_count = $4,
          download_hash_type = $5,
          download_hash_value = $6,
          download_url_host = $7,
          reconciliation_batch_id = $8::uuid,
          completed_at = now(),
          failed_at = null,
          error_code = null,
          error_message = null,
          metadata = metadata || $9::jsonb,
          updated_at = now()
      where tenant_id = $1 and id = $2
      `,
      [
        job.tenantId,
        job.id,
        downloaded.sourceHash,
        downloaded.rows.length,
        downloaded.downloadHashType,
        downloaded.downloadHashValue,
        downloaded.downloadUrlHost,
        imported.batch.id,
        JSON.stringify({
          completedBy: config.providerBillWorkerId,
          completedAt: new Date().toISOString(),
          reconciliationBatchId: imported.batch.id,
          reconciliationStatus: imported.batch.status,
          reconciliationSummary: {
            totalCount: imported.batch.totalCount,
            matchedCount: imported.batch.matchedCount,
            mismatchCount: imported.batch.mismatchCount,
            missingLocalCount: imported.batch.missingLocalCount,
            missingProviderCount: imported.batch.missingProviderCount,
            duplicateCount: imported.batch.duplicateCount,
          },
        }),
      ],
    );
    await client.query('commit');
    return 'completed';
  } catch (error) {
    await client.query('rollback').catch(() => {});
    const failure = publicFailure(error);
    await client.query(
      `
      update public.commerce_bill_download_jobs
      set status = 'failed',
          failed_at = now(),
          error_code = $3,
          error_message = $4,
          metadata = metadata || $5::jsonb,
          updated_at = now()
      where tenant_id = $1 and id = $2
      `,
      [
        job.tenantId,
        job.id,
        failure.code,
        failure.message,
        JSON.stringify({
          failedBy: config.providerBillWorkerId,
          failedAt: new Date().toISOString(),
        }),
      ],
    ).catch(() => {});
    return 'failed';
  } finally {
    client.release();
  }
}

function publicFailure(error: unknown) {
  const code = error instanceof ProviderBillWorkerError ? error.code : 'PROVIDER_BILL_WORKER_ERROR';
  const message = error instanceof Error ? error.message : 'Provider bill worker failed';
  return {
    code,
    message: message
      .replace(/-----BEGIN[\s\S]+?-----END [A-Z ]+-----/g, '[redacted-pem]')
      .slice(0, 500),
  };
}

export async function processProviderBillBatch(limit = config.providerBillBatchSize): Promise<ProviderBillWorkerResult> {
  const client = await pool.connect();
  let jobs: ProviderBillJob[] = [];
  try {
    await client.query('begin');
    jobs = await claimProviderBillJobs(client, limit);
    await client.query('commit');
  } catch (error) {
    await client.query('rollback');
    throw error;
  } finally {
    client.release();
  }

  const result: ProviderBillWorkerResult = {
    processed: jobs.length,
    completed: 0,
    failed: 0,
    skipped: 0,
  };

  for (const job of jobs) {
    const status = await processProviderBillJob(job);
    if (status === 'completed') result.completed += 1;
    else if (status === 'failed') result.failed += 1;
    else result.skipped += 1;
  }
  return result;
}
