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

const databaseUrl = process.env.DATABASE_URL || 'postgresql://postgres:postgres@127.0.0.1:54322/postgres';
const tenantId = '00000000-0000-0000-0000-000000000001';

const ids = {
  okAsset: '20000000-0000-0000-0000-000000000801',
  badAsset: '20000000-0000-0000-0000-000000000802',
  scanBadAsset: '20000000-0000-0000-0000-000000000803',
  httpPassAsset: '20000000-0000-0000-0000-000000000804',
  httpFailAsset: '20000000-0000-0000-0000-000000000805',
  httpUnavailableAsset: '20000000-0000-0000-0000-000000000806',
};

const checksumA = 'a'.repeat(64);
const checksumB = 'b'.repeat(64);

async function runWorkerOnce(extraEnv = {}) {
  const child = spawn(process.execPath, ['apps/worker/dist/apps/worker/src/index.js', '--once', '--job', 'assets'], {
    cwd: process.cwd(),
    env: {
      ...process.env,
      DATABASE_URL: databaseUrl,
      WORKER_ASSET_BATCH_SIZE: '20',
      WORKER_ASSET_MIN_AGE_SECONDS: '0',
      WORKER_ASSET_RECHECK_INTERVAL_SECONDS: '86400',
      WORKER_ASSET_REQUEST_TIMEOUT_MS: '5000',
      WORKER_ASSET_SECURITY_SCANNER: 'metadata_rules',
      WORKER_ASSET_SECURITY_SCAN_HTTP_ENDPOINT: '',
      WORKER_ASSET_SECURITY_SCAN_HTTP_TOKEN: '',
      WORKER_ASSET_SECURITY_SCAN_HTTP_TIMEOUT_MS: '10000',
      WORKER_ASSET_SECURITY_SCAN_FAIL_OPEN: 'false',
      STORAGE_REQUIRE_TENANT_PREFIX: 'true',
      ...extraEnv,
    },
    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, /assets batch processed=\d+/, 'worker output should include assets summary');
  return output;
}

async function startScannerServer() {
  const requests = [];
  const server = http.createServer((request, response) => {
    let body = '';
    request.on('data', chunk => {
      body += chunk.toString();
    });
    request.on('end', () => {
      const payload = JSON.parse(body || '{}');
      requests.push({
        authorization: request.headers.authorization,
        payload,
      });
      response.setHeader('content-type', 'application/json');
      if (payload.assetId === ids.httpFailAsset) {
        response.end(JSON.stringify({
          status: 'failed',
          riskLevel: 'critical',
          issueCodes: ['virus_detected'],
          details: {
            engine: 'fake-av',
            signature: 'EICAR-Test-File',
            accessToken: 'must-be-redacted',
          },
        }));
        return;
      }
      response.end(JSON.stringify({
        status: 'passed',
        riskLevel: 'none',
        issueCodes: [],
        details: {
          engine: 'fake-av',
          scannedBytes: payload.fileSizeBytes,
        },
      }));
    });
  });
  await new Promise(resolve => server.listen(0, '127.0.0.1', resolve));
  const address = server.address();
  assert.ok(address && typeof address === 'object', 'scanner server should listen on a TCP port');
  return {
    endpoint: `http://127.0.0.1:${address.port}/scan`,
    requests,
    close: () => new Promise(resolve => server.close(resolve)),
  };
}

function countFromWorkerOutput(output, name) {
  const match = output.match(new RegExp(`${name}=(\\d+)`));
  return match ? Number(match[1]) : 0;
}

async function cleanup(pool) {
  await pool.query(
    `
    delete from public.audit_logs
    where tenant_id = $1
      and target_type = 'content_asset'
      and target_id = any($2::text[])
    `,
    [tenantId, Object.values(ids)],
  );
  await pool.query(
    `
    delete from public.content_asset_security_scan_events
    where tenant_id = $1 and asset_id = any($2::uuid[])
    `,
    [tenantId, Object.values(ids)],
  );
  await pool.query(
    `
    delete from public.content_assets
    where tenant_id = $1 and id = any($2::uuid[])
    `,
    [tenantId, Object.values(ids)],
  );
}

async function seed(pool) {
  await pool.query(
    `
    insert into public.content_assets (
      id, tenant_id, asset_key, title, asset_type, storage_provider,
      bucket, object_key, file_name, mime_type, file_size_bytes, checksum_sha256,
      visibility, status, upload_status, verified_at, verified_size_bytes,
      verified_checksum_sha256, verification_details,
      security_scan_status, security_scan_provider, security_scan_summary,
      security_flags, source,
      created_at, updated_at
    )
    values
      (
        $1, $2, 'asset-worker-ok', '资源复检正常 PDF', 'pdf', 'local_dev',
        'tenant-assets', $3, 'ok.pdf', 'application/pdf', 4096, $4,
        'tenant', 'active', 'verified', now() - interval '2 days', 4096,
        $4, '{"source":"asset-worker-test"}'::jsonb,
        'pending', 'metadata_rules', '{}'::jsonb,
        '{}'::jsonb, 'integration-test',
        now() - interval '2 days', now() - interval '2 days'
      ),
      (
        $5, $2, 'asset-worker-bad', '资源复检异常 PDF', 'pdf', 'local_dev',
        'tenant-assets', $6, 'bad.pdf', 'application/pdf', 1024, $7,
        'tenant', 'active', 'verified', now() - interval '2 days', 2048,
        $7, '{"source":"asset-worker-test"}'::jsonb,
        'pending', 'metadata_rules', '{}'::jsonb,
        '{}'::jsonb, 'integration-test',
        now() - interval '2 days', now() - interval '2 days'
      ),
      (
        $8, $2, 'asset-worker-scan-bad', '资源安全扫描异常 PDF', 'pdf', 'local_dev',
        'tenant-assets', $9, 'scan-bad.pdf', 'application/pdf', 4096, $4,
        'tenant', 'active', 'verified', now() - interval '2 days', 4096,
        $4, '{"source":"asset-worker-test"}'::jsonb,
        'pending', 'metadata_rules', '{"securityScanForceFail":true}'::jsonb,
        '{}'::jsonb, 'integration-test',
        now() - interval '2 days', now() - interval '2 days'
      ),
      (
        $10, $2, 'asset-worker-http-pass', '外部扫描正常 PDF', 'pdf', 'local_dev',
        'tenant-assets', $11, 'http-pass.pdf', 'application/pdf', 4096, $4,
        'tenant', 'active', 'verified', now() - interval '2 days', 4096,
        $4, '{"source":"asset-worker-test"}'::jsonb,
        'pending', 'metadata_rules', '{}'::jsonb,
        '{}'::jsonb, 'integration-test',
        now() - interval '2 days', now() - interval '2 days'
      ),
      (
        $12, $2, 'asset-worker-http-fail', '外部扫描异常 PDF', 'pdf', 'local_dev',
        'tenant-assets', $13, 'http-fail.pdf', 'application/pdf', 4096, $4,
        'tenant', 'active', 'verified', now() - interval '2 days', 4096,
        $4, '{"source":"asset-worker-test"}'::jsonb,
        'pending', 'metadata_rules', '{}'::jsonb,
        '{}'::jsonb, 'integration-test',
        now() - interval '2 days', now() - interval '2 days'
      ),
      (
        $14, $2, 'asset-worker-http-unavailable', '外部扫描不可用 PDF', 'pdf', 'local_dev',
        'tenant-assets', $15, 'http-unavailable.pdf', 'application/pdf', 4096, $4,
        'tenant', 'active', 'verified', now() - interval '2 days', 4096,
        $4, '{"source":"asset-worker-test"}'::jsonb,
        'pending', 'metadata_rules', '{}'::jsonb,
        '{}'::jsonb, 'integration-test',
        now() - interval '2 days', now() - interval '2 days'
      )
    `,
    [
      ids.okAsset,
      tenantId,
      `${tenantId}/assets/worker-ok.pdf`,
      checksumA,
      ids.badAsset,
      `${tenantId}/assets/worker-bad.pdf`,
      checksumB,
      ids.scanBadAsset,
      `${tenantId}/assets/worker-scan-bad.pdf`,
      ids.httpPassAsset,
      `${tenantId}/assets/worker-http-pass.pdf`,
      ids.httpFailAsset,
      `${tenantId}/assets/worker-http-fail.pdf`,
      ids.httpUnavailableAsset,
      `${tenantId}/assets/worker-http-unavailable.pdf`,
    ],
  );
}

async function resetAssetsForHttp(pool) {
  await pool.query(
    `
    update public.content_assets
    set status = 'active',
        upload_status = 'verified',
        verified_at = now() - interval '2 days',
        verification_details = '{"source":"asset-worker-test"}'::jsonb,
        security_scan_status = 'pending',
        security_scan_provider = 'metadata_rules',
        security_scan_summary = '{}'::jsonb,
        security_flags = '{}'::jsonb,
        updated_at = now() - interval '2 days'
    where tenant_id = $1
      and id = any($2::uuid[])
    `,
    [tenantId, [ids.httpPassAsset, ids.httpFailAsset, ids.httpUnavailableAsset]],
  );
  await pool.query(
    `
    delete from public.content_asset_security_scan_events
    where tenant_id = $1 and asset_id = any($2::uuid[])
    `,
    [tenantId, [ids.httpPassAsset, ids.httpFailAsset, ids.httpUnavailableAsset]],
  );
}

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

    const output = await runWorkerOnce();
    assert.ok(countFromWorkerOutput(output, 'verified') >= 3, 'worker should verify metadata-safe assets');
    assert.ok(countFromWorkerOutput(output, 'failed') >= 2, 'worker should fail metadata and security scan assets');

    const assets = await pool.query(
      `
      select id, status, upload_status, verification_details,
             security_scan_status, security_scan_provider, security_scan_summary,
             security_flags
      from public.content_assets
      where tenant_id = $1 and id = any($2::uuid[])
      order by id
      `,
      [tenantId, [ids.okAsset, ids.badAsset, ids.scanBadAsset, ids.httpPassAsset, ids.httpFailAsset, ids.httpUnavailableAsset]],
    );
    const okAsset = assets.rows.find(row => row.id === ids.okAsset);
    const badAsset = assets.rows.find(row => row.id === ids.badAsset);
    const scanBadAsset = assets.rows.find(row => row.id === ids.scanBadAsset);
    assert.equal(okAsset?.status, 'active', 'verified asset should remain active');
    assert.equal(okAsset?.upload_status, 'verified', 'verified asset should remain verified');
    assert.equal(okAsset?.security_scan_status, 'passed', 'verified asset should pass security scan');
    assert.equal(okAsset?.security_scan_provider, 'metadata_rules', 'verified asset should record scan provider');
    assert.equal(okAsset?.verification_details?.assetWorker?.lastResult, 'verified', 'verified asset should record worker result');
    assert.equal(okAsset?.security_flags?.assetRecheckFailed, undefined, 'verified asset should not keep recheck failure flag');

    assert.equal(badAsset?.status, 'draft', 'mismatched asset should be unpublished');
    assert.equal(badAsset?.upload_status, 'failed', 'mismatched asset should be marked failed');
    assert.equal(badAsset?.security_scan_status, 'skipped', 'mismatched asset should skip security scan');
    assert.equal(badAsset?.security_flags?.assetRecheckFailed, true, 'mismatched asset should record security flag');
    assert.deepEqual(
      badAsset?.verification_details?.assetWorker?.issues,
      ['file_size_mismatch'],
      'mismatched asset should record exact issue',
    );
    assert.equal(scanBadAsset?.status, 'draft', 'security failed asset should be unpublished');
    assert.equal(scanBadAsset?.upload_status, 'verified', 'security failed asset should keep upload verification evidence');
    assert.equal(scanBadAsset?.security_scan_status, 'failed', 'security failed asset should record failed scan status');
    assert.equal(scanBadAsset?.security_flags?.assetSecurityScanFailed, true, 'security failed asset should record security flag');
    assert.ok(
      scanBadAsset?.security_scan_summary?.issueCodes?.includes('security_scan_forced_failure'),
      'security failed asset should record issue code',
    );

    const scanEvents = await pool.query(
      `
      select asset_id, scan_status, risk_level, issue_codes, provider, details
      from public.content_asset_security_scan_events
      where tenant_id = $1 and asset_id in ($2::uuid, $3::uuid)
      order by created_at asc
      `,
      [tenantId, ids.okAsset, ids.scanBadAsset],
    );
    assert.ok(
      scanEvents.rows.some(row => row.asset_id === ids.okAsset && row.scan_status === 'passed' && row.provider === 'metadata_rules'),
      'worker should write passed security scan event',
    );
    assert.ok(
      scanEvents.rows.some(row => row.asset_id === ids.scanBadAsset && row.scan_status === 'failed' && row.issue_codes.includes('security_scan_forced_failure')),
      'worker should write failed security scan event',
    );

    const skippedScanEvents = await pool.query(
      `
      select asset_id, scan_status, risk_level, issue_codes, provider, details
      from public.content_asset_security_scan_events
      where tenant_id = $1 and asset_id = $2::uuid
      order by created_at asc
      `,
      [tenantId, ids.badAsset],
    );
    assert.ok(
      skippedScanEvents.rows.some(row => row.scan_status === 'skipped' && row.issue_codes.includes('file_size_mismatch')),
      'worker should write skipped scan event when metadata verification fails',
    );

    const audits = await pool.query(
      `
      select action, details
      from public.audit_logs
      where tenant_id = $1
        and target_type = 'content_asset'
        and target_id in ($2, $3, $4)
      order by created_at asc
      `,
      [tenantId, ids.okAsset, ids.badAsset, ids.scanBadAsset],
    );
    assert.ok(
      audits.rows.some(row => row.action === 'content.asset.rechecked' && row.details?.result === 'verified'),
      'worker should write verified audit log',
    );
    assert.ok(
      audits.rows.some(row => row.action === 'content.asset.recheck_failed' && row.details?.issues?.includes('file_size_mismatch')),
      'worker should write failed audit log',
    );
    assert.ok(
      audits.rows.some(row => row.action === 'content.asset.security_scan_failed' && row.details?.issueCodes?.includes('security_scan_forced_failure')),
      'worker should write failed security scan audit log',
    );

    await resetAssetsForHttp(pool);
    const scanner = await startScannerServer();
    try {
      const httpOutput = await runWorkerOnce({
        WORKER_ASSET_SECURITY_SCANNER: 'metadata_rules,http',
        WORKER_ASSET_SECURITY_SCAN_HTTP_ENDPOINT: scanner.endpoint,
        WORKER_ASSET_SECURITY_SCAN_HTTP_TOKEN: 'scanner-test-token',
        WORKER_ASSET_SECURITY_SCAN_HTTP_TIMEOUT_MS: '5000',
        WORKER_ASSET_SECURITY_SCAN_FAIL_OPEN: 'false',
      });
      assert.ok(countFromWorkerOutput(httpOutput, 'verified') >= 1, 'worker should pass externally clean assets');
      assert.ok(countFromWorkerOutput(httpOutput, 'failed') >= 1, 'worker should fail externally rejected assets');
      assert.equal(scanner.requests.length, 3, 'external scanner should receive metadata-clean assets only');
      assert.ok(
        scanner.requests.every(item => item.authorization === 'Bearer scanner-test-token'),
        'external scanner should receive bearer token',
      );

      const httpAssets = await pool.query(
        `
        select id, status, security_scan_status, security_scan_provider, security_scan_summary, security_flags
        from public.content_assets
        where tenant_id = $1 and id = any($2::uuid[])
        order by id
        `,
        [tenantId, [ids.httpPassAsset, ids.httpFailAsset, ids.httpUnavailableAsset]],
      );
      const httpPassAsset = httpAssets.rows.find(row => row.id === ids.httpPassAsset);
      const httpFailAsset = httpAssets.rows.find(row => row.id === ids.httpFailAsset);
      const httpUnavailableAsset = httpAssets.rows.find(row => row.id === ids.httpUnavailableAsset);

      assert.equal(httpPassAsset?.status, 'active', 'external scan pass asset should stay active');
      assert.equal(httpPassAsset?.security_scan_status, 'passed', 'external scan pass asset should pass');
      assert.equal(
        httpPassAsset?.security_scan_provider,
        'metadata_rules+http_security_scan',
        'external scan pass asset should record merged provider',
      );
      assert.equal(httpFailAsset?.status, 'draft', 'external scan failed asset should be unpublished');
      assert.equal(httpFailAsset?.security_scan_status, 'failed', 'external scan failed asset should fail');
      assert.ok(
        httpFailAsset?.security_scan_summary?.issueCodes?.includes('virus_detected'),
        'external scan failed asset should include provider issue code',
      );
      const scannerDetails = httpFailAsset?.security_scan_summary?.details?.results?.find(
        result => result.provider === 'http_security_scan',
      )?.details?.scannerDetails;
      assert.equal(scannerDetails?.accessToken, '[redacted]', 'scanner details should redact secret-like keys');
      assert.equal(httpUnavailableAsset?.security_scan_status, 'passed', 'available scanner should pass clean third asset');

      const httpScanEvents = await pool.query(
        `
        select asset_id, scan_status, provider, issue_codes, details
        from public.content_asset_security_scan_events
        where tenant_id = $1 and asset_id = any($2::uuid[])
        order by created_at asc
        `,
        [tenantId, [ids.httpPassAsset, ids.httpFailAsset]],
      );
      assert.ok(
        httpScanEvents.rows.some(row => row.asset_id === ids.httpPassAsset && row.provider === 'metadata_rules+http_security_scan' && row.scan_status === 'passed'),
        'worker should write external pass event',
      );
      assert.ok(
        httpScanEvents.rows.some(row => row.asset_id === ids.httpFailAsset && row.issue_codes.includes('virus_detected')),
        'worker should write external failure event',
      );
    } finally {
      await scanner.close();
    }

    await resetAssetsForHttp(pool);
    const unavailableOutput = await runWorkerOnce({
      WORKER_ASSET_SECURITY_SCANNER: 'metadata_rules,http',
      WORKER_ASSET_SECURITY_SCAN_HTTP_ENDPOINT: 'http://127.0.0.1:9/scan',
      WORKER_ASSET_SECURITY_SCAN_HTTP_TOKEN: 'scanner-test-token',
      WORKER_ASSET_SECURITY_SCAN_HTTP_TIMEOUT_MS: '300',
      WORKER_ASSET_SECURITY_SCAN_FAIL_OPEN: 'false',
    });
    assert.ok(countFromWorkerOutput(unavailableOutput, 'failed') >= 3, 'fail-closed scanner outage should fail assets');
    const unavailableAssets = await pool.query(
      `
      select id, status, security_scan_status, security_scan_summary, security_flags
      from public.content_assets
      where tenant_id = $1 and id = any($2::uuid[])
      `,
      [tenantId, [ids.httpPassAsset, ids.httpFailAsset, ids.httpUnavailableAsset]],
    );
    for (const row of unavailableAssets.rows) {
      assert.equal(row.status, 'draft', 'fail-closed scanner outage should unpublish active asset');
      assert.equal(row.security_scan_status, 'failed', 'fail-closed scanner outage should fail scan');
      assert.ok(
        row.security_scan_summary?.issueCodes?.includes('external_security_scan_unavailable'),
        'fail-closed scanner outage should record unavailable issue',
      );
      assert.equal(row.security_flags?.assetSecurityScanFailed, true, 'fail-closed scanner outage should record security flag');
    }

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

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