import assert from 'node:assert/strict';
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 ids = {
  region: '00000000-0000-0000-0000-000000000301',
  subject: '00000000-0000-0000-0000-000000000501',
  category: '00000000-0000-0000-0000-000000000601',
  contentEntry: '00000000-0000-0000-0000-000000000611',
  contentNodeSchoolTarget: '00000000-0000-0000-0000-000000000614',
  questionCollection: '00000000-0000-0000-0000-000000000615',
};

function runWorkerOnce() {
  const child = spawn(process.execPath, ['apps/worker/dist/apps/worker/src/index.js', '--once', '--job', 'imports'], {
    cwd: process.cwd(),
    env: {
      ...process.env,
      DATABASE_URL: databaseUrl,
      WORKER_IMPORT_BATCH_SIZE: '5',
      WORKER_IMPORT_ID: 'imports-integration-test',
    },
    stdio: ['ignore', 'pipe', 'pipe'],
    windowsHide: true,
  });
  let output = '';
  child.stdout.on('data', chunk => {
    output += chunk.toString();
  });
  child.stderr.on('data', chunk => {
    output += chunk.toString();
  });
  return new Promise((resolve, reject) => {
    child.on('error', reject);
    child.on('exit', code => {
      try {
        assert.equal(code, 0, `worker should exit 0\n${output}`);
        assert.match(output, /imports batch processed=\d+/, 'worker output should include imports summary');
        resolve(output);
      } catch (error) {
        reject(error);
      }
    });
  });
}

async function cleanup(pool) {
  await pool.query(
    `
    delete from public.question_collection_items
    where tenant_id = $1
      and question_id in (
        select id from public.questions
        where tenant_id = $1
          and (
            legacy_id like 'worker-import-question-%'
            or legacy_id like 'integration-import-async-choice-%'
          )
      )
    `,
    [tenantId],
  );
  await pool.query(
    `
    delete from public.question_versions
    where tenant_id = $1
      and question_id in (
        select id from public.questions
        where tenant_id = $1
          and (
            legacy_id like 'worker-import-question-%'
            or legacy_id like 'integration-import-async-choice-%'
          )
      )
    `,
    [tenantId],
  );
  await pool.query(
    `
    delete from public.questions
    where tenant_id = $1
      and (
        legacy_id like 'worker-import-question-%'
        or legacy_id like 'integration-import-async-choice-%'
      )
    `,
    [tenantId],
  );
  await pool.query(
    `
    delete from public.audit_logs
    where tenant_id = $1
      and target_type = 'content_import_job'
      and details::text like '%worker-import%'
    `,
    [tenantId],
  );
  await pool.query(
    `
    delete from public.content_import_jobs
    where tenant_id = $1
      and (
        source_name like 'worker-import-%'
        or source_name = 'async-question-import.json'
      )
    `,
    [tenantId],
  );
}

async function createQueuedQuestionImport(pool) {
  const preview = await pool.query(
    `
    insert into public.content_import_jobs (
      tenant_id, created_by, import_type, source_format, status,
      source_name, source_hash, target_region_id, target_subject_id,
      target_category_id, target_content_node_id, target_collection_id,
      dry_run, total_count, valid_count, error_count, warning_count,
      summary, raw_payload, normalized_payload, execution_mode, queued_at,
      next_attempt_at, parser_metadata
    )
    values (
      $1, $2, 'questions', 'json', 'pending',
      'worker-import-questions.json', 'worker-import-source-hash',
      $3::uuid, $4::uuid, $5::uuid, $6::uuid, $7::uuid,
      false, 1, 1, 0, 0,
      $8::jsonb, $9::jsonb, $10::jsonb, 'async', now(), now(), '{}'::jsonb
    )
    returning id
    `,
    [
      tenantId,
      adminUserId,
      ids.region,
      ids.subject,
      ids.category,
      ids.contentNodeSchoolTarget,
      ids.questionCollection,
      JSON.stringify({
        target: {
          regionId: ids.region,
          subjectId: ids.subject,
          categoryId: ids.category,
          contentNodeId: ids.contentNodeSchoolTarget,
          collectionId: ids.questionCollection,
        },
        importOptions: { allowPartial: false },
        source: 'worker-import-integration',
      }),
      JSON.stringify([
        {
          legacyId: 'worker-import-question-001',
          type: 'choice',
          content: '异步导入题：worker 应该复用哪套导入规则？',
          options: ['自己重写', '复用后端导入 executor', '前端直写数据库', '跳过校验'],
          correctOptionIndices: [1],
          explanation: 'worker 和 API 必须复用同一套后端导入规则。',
          difficulty: 2,
          tags: ['worker-import'],
        },
      ]),
      JSON.stringify([
        {
          legacyId: 'worker-import-question-001',
          type: 'choice',
          typeLabel: null,
          content: '异步导入题：worker 应该复用哪套导入规则？',
          options: ['自己重写', '复用后端导入 executor', '前端直写数据库', '跳过校验'],
          correctOptionIndex: 1,
          correctOptionIndices: [1],
          answerText: null,
          explanation: 'worker 和 API 必须复用同一套后端导入规则。',
          difficulty: 2,
          tags: ['worker-import'],
          mediaUrl: null,
          subQuestions: [],
          codeLang: null,
          codeTemplate: null,
          examMarkers: {},
          sourceHash: 'worker-import-question-hash-001',
        },
      ]),
    ],
  );
  const jobId = preview.rows[0].id;
  await pool.query(
    `
    insert into public.content_import_items (
      tenant_id, job_id, row_no, external_id, status, target_type,
      source_payload, normalized_payload, content_hash, issues_count
    )
    values ($1, $2, 1, 'worker-import-question-001', 'valid', 'question', $3::jsonb, $4::jsonb, 'worker-import-question-hash-001', 0)
    `,
    [
      tenantId,
      jobId,
      JSON.stringify({
        legacyId: 'worker-import-question-001',
        content: '异步导入题：worker 应该复用哪套导入规则？',
      }),
      JSON.stringify({
        legacyId: 'worker-import-question-001',
        type: 'choice',
        typeLabel: null,
        content: '异步导入题：worker 应该复用哪套导入规则？',
        options: ['自己重写', '复用后端导入 executor', '前端直写数据库', '跳过校验'],
        correctOptionIndex: 1,
        correctOptionIndices: [1],
        answerText: null,
        explanation: 'worker 和 API 必须复用同一套后端导入规则。',
        difficulty: 2,
        tags: ['worker-import'],
        mediaUrl: null,
        subQuestions: [],
        codeLang: null,
        codeTemplate: null,
        examMarkers: {},
        sourceHash: 'worker-import-question-hash-001',
      }),
    ],
  );
  return jobId;
}

async function main() {
  const pool = new pg.Pool({ connectionString: databaseUrl });
  try {
    await cleanup(pool);
    const jobId = await createQueuedQuestionImport(pool);

    const output = await runWorkerOnce();
    assert.match(output, /completed=1/, 'worker should complete exactly the queued import job after cleanup');

    const job = await pool.query(
      `
      select status, execution_mode, inserted_count, updated_count, skipped_count,
             locked_at, locked_by, attempt_count, error_message
      from public.content_import_jobs
      where tenant_id = $1 and id = $2
      `,
      [tenantId, jobId],
    );
    assert.equal(job.rows[0]?.status, 'completed', 'queued import job should be completed');
    assert.equal(job.rows[0]?.execution_mode, 'async', 'job should keep async execution mode');
    assert.equal(Number(job.rows[0]?.inserted_count), 1, 'worker should insert one question');
    assert.equal(job.rows[0]?.locked_at, null, 'completed job should release lock');
    assert.equal(job.rows[0]?.locked_by, null, 'completed job should clear lock owner');
    assert.equal(Number(job.rows[0]?.attempt_count), 1, 'worker should record one attempt');
    assert.equal(job.rows[0]?.error_message, null, 'completed job should not retain error message');

    const question = await pool.query(
      `
      select q.id, v.content
      from public.questions q
      join public.question_versions v on v.id = q.current_version_id
      where q.tenant_id = $1 and q.legacy_id = 'worker-import-question-001'
      limit 1
      `,
      [tenantId],
    );
    assert.equal(question.rows[0]?.content, '异步导入题：worker 应该复用哪套导入规则？', 'worker should import question content');

    const collectionItem = await pool.query(
      `
      select 1
      from public.question_collection_items
      where tenant_id = $1 and collection_id = $2 and question_id = $3
      limit 1
      `,
      [tenantId, ids.questionCollection, question.rows[0]?.id],
    );
    assert.equal(collectionItem.rowCount, 1, 'worker should bind imported question to collection');

    const audit = await pool.query(
      `
      select action
      from public.audit_logs
      where tenant_id = $1 and target_type = 'content_import_job' and target_id = $2
      order by created_at desc
      limit 1
      `,
      [tenantId, jobId],
    );
    assert.equal(audit.rows[0]?.action, 'content.import.questions.completed', 'worker import should write completion audit');

    console.log('Import worker integration test complete.');
  } finally {
    await cleanup(pool).catch(() => {});
    await pool.end();
  }
}

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