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

import assert from 'node:assert/strict';
import { fileURLToPath, pathToFileURL } from 'node:url';
import pg from 'pg';
import { setTimeout as delay } from 'node:timers/promises';
import {
  assertDestructiveTestDatabase,
  resolveDestructiveTestConfirmation,
} from './lib/destructive-test-database-guard.js';

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 confirmation = resolveDestructiveTestConfirmation();
const ids = {
  region: '00000000-0000-0000-0000-000000000301',
  subject: '00000000-0000-0000-0000-000000000501',
  category: '00000000-0000-0000-0000-000000000601',
  contentNodeSchoolTarget: '00000000-0000-0000-0000-000000000614',
  questionCollection: '00000000-0000-0000-0000-000000000615',
};

process.env.DATABASE_URL = databaseUrl;
process.env.WORKER_IMPORT_BATCH_SIZE = '5';
process.env.WORKER_IMPORT_ID = 'imports-integration-default';
process.env.WORKER_IMPORT_LEASE_SECONDS = '120';
process.env.WORKER_IMPORT_HEARTBEAT_INTERVAL_MS = '30000';

const workerModuleUrl = pathToFileURL(fileURLToPath(new URL('../apps/worker/dist/apps/worker/src/jobs/imports.js', import.meta.url))).href;
const apiModuleUrl = pathToFileURL(fileURLToPath(new URL('../apps/worker/dist/apps/api/src/features/tenant-content/imports.js', import.meta.url))).href;
const worker = await import(workerModuleUrl);
const { executeContentImportJob } = await import(apiModuleUrl);

function auth() {
  return {
    tenantId,
    userId: adminUserId,
    role: 'system_worker',
    permissions: { 'content:*': true },
    templatePermissions: {},
  };
}

async function cleanup(pool) {
  await pool.query(
    `
    update public.content_import_jobs
    set status = 'pending',
        locked_at = null,
        locked_by = null,
        lease_token = null,
        lease_expires_at = null,
        last_heartbeat_at = null,
        next_attempt_at = now(),
        updated_at = now()
    where tenant_id = $1
      and source_name like 'worker-lease-%'
      and status = 'importing'
    `,
    [tenantId],
  );
  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-lease-question-%'
      )
    `,
    [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-lease-question-%'
      )
    `,
    [tenantId],
  );
  await pool.query(
    `delete from public.questions where tenant_id = $1 and legacy_id like 'worker-lease-question-%'`,
    [tenantId],
  );
  await pool.query(
    `
    delete from public.audit_logs
    where tenant_id = $1
      and target_type = 'content_import_job'
      and target_id in (
        select id::text from public.content_import_jobs
        where tenant_id = $1 and source_name like 'worker-lease-%'
      )
    `,
    [tenantId],
  );
  await pool.query(
    `delete from public.content_import_jobs where tenant_id = $1 and source_name like 'worker-lease-%'`,
    [tenantId],
  );
}

async function createQueuedQuestionImport(pool, suffix, maxAttempts = 3) {
  const legacyId = `worker-lease-question-${suffix}`;
  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, max_attempts, parser_metadata
    )
    values (
      $1, $2, 'questions', 'json', 'pending',
      $3, $4, $5::uuid, $6::uuid, $7::uuid, $8::uuid, $9::uuid,
      false, 1, 1, 0, 0,
      $10::jsonb, $11::jsonb, $12::jsonb, 'async', now(), now(), $13, '{}'::jsonb
    )
    returning id
    `,
    [
      tenantId,
      adminUserId,
      `worker-lease-${suffix}.json`,
      `worker-lease-source-${suffix}`,
      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-lease-integration',
      }),
      JSON.stringify([{ legacyId, type: 'choice', content: `lease test ${suffix}` }]),
      JSON.stringify([{
        legacyId,
        type: 'choice',
        typeLabel: null,
        content: `lease test ${suffix}`,
        options: ['A', 'B'],
        correctOptionIndex: 1,
        correctOptionIndices: [1],
        answerText: null,
        explanation: 'persistent lease integration fixture',
        difficulty: 2,
        tags: ['worker-lease'],
        mediaUrl: null,
        subQuestions: [],
        codeLang: null,
        codeTemplate: null,
        examMarkers: {},
        sourceHash: `worker-lease-hash-${suffix}`,
      }]),
      maxAttempts,
    ],
  );
  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, $3, 'valid', 'question', $4::jsonb, $5::jsonb, $6, 0)
    `,
    [
      tenantId,
      jobId,
      legacyId,
      JSON.stringify({ legacyId, content: `lease test ${suffix}` }),
      JSON.stringify({
        legacyId,
        type: 'choice',
        typeLabel: null,
        content: `lease test ${suffix}`,
        options: ['A', 'B'],
        correctOptionIndex: 1,
        correctOptionIndices: [1],
        answerText: null,
        explanation: 'persistent lease integration fixture',
        difficulty: 2,
        tags: ['worker-lease'],
        mediaUrl: null,
        subQuestions: [],
        codeLang: null,
        codeTemplate: null,
        examMarkers: {},
        sourceHash: `worker-lease-hash-${suffix}`,
      }),
      `worker-lease-hash-${suffix}`,
    ],
  );
  return { jobId, legacyId };
}

async function readJob(pool, jobId) {
  const result = await pool.query(
    `
    select status, attempt_count as "attemptCount", locked_by as "lockedBy",
           lease_token as "leaseToken", lease_expires_at as "leaseExpiresAt",
           last_heartbeat_at as "lastHeartbeatAt", next_attempt_at as "nextAttemptAt",
           inserted_count as "insertedCount", error_message as "errorMessage"
    from public.content_import_jobs
    where tenant_id = $1 and id = $2
    `,
    [tenantId, jobId],
  );
  return result.rows[0];
}

async function testAtomicClaim(pool) {
  const jobs = await Promise.all([
    createQueuedQuestionImport(pool, 'atomic-a'),
    createQueuedQuestionImport(pool, 'atomic-b'),
    createQueuedQuestionImport(pool, 'atomic-c'),
  ]);
  const [left, right] = await Promise.all([
    worker.claimImportJobs({ workerId: 'lease-worker-a', batchSize: 2, leaseSeconds: 30 }),
    worker.claimImportJobs({ workerId: 'lease-worker-b', batchSize: 2, leaseSeconds: 30 }),
  ]);
  const claimed = [...left, ...right];
  assert.equal(claimed.length, 3, 'concurrent workers should claim all three jobs');
  assert.equal(new Set(claimed.map(job => job.id)).size, 3, 'SKIP LOCKED claim must not duplicate a job');
  assert.deepEqual(
    new Set(claimed.map(job => job.id)),
    new Set(jobs.map(job => job.jobId)),
    'claims must stay within the ready fixture set',
  );
  for (const job of claimed) {
    assert.equal(job.attemptCount, 1, 'claim should atomically increment attempt_count once');
    assert.ok(job.leaseToken, 'claim should persist a fencing token');
  }
}

async function testHeartbeat(pool) {
  const fixture = await createQueuedQuestionImport(pool, 'heartbeat');
  const [claimed] = await worker.claimImportJobs({
    workerId: 'lease-heartbeat-worker',
    batchSize: 1,
    leaseSeconds: 3,
  });
  assert.equal(claimed.id, fixture.jobId);
  const before = await readJob(pool, fixture.jobId);
  const heartbeat = worker.startImportLeaseHeartbeat(claimed, {
    leaseSeconds: 3,
    heartbeatIntervalMs: 250,
  });
  await delay(700);
  await heartbeat.stop();
  const after = await readJob(pool, fixture.jobId);
  assert.ok(after.lastHeartbeatAt > before.lastHeartbeatAt, 'heartbeat should advance last_heartbeat_at');
  assert.ok(after.leaseExpiresAt > before.leaseExpiresAt, 'heartbeat should extend lease expiry');
}

async function testExpiredTakeoverAndFencing(pool) {
  const fixture = await createQueuedQuestionImport(pool, 'takeover');
  const [first] = await worker.claimImportJobs({ workerId: 'lease-old-worker', batchSize: 1, leaseSeconds: 30 });
  await pool.query(
    `
    update public.content_import_jobs
    set locked_at = now() - interval '10 seconds',
        lease_expires_at = now() - interval '1 second',
        last_heartbeat_at = now() - interval '10 seconds'
    where tenant_id = $1 and id = $2
    `,
    [tenantId, fixture.jobId],
  );

  const [second] = await worker.claimImportJobs({ workerId: 'lease-new-worker', batchSize: 1, leaseSeconds: 30 });
  assert.equal(second.id, fixture.jobId, 'expired importing job should be reclaimed');
  assert.notEqual(second.leaseToken, first.leaseToken, 'takeover must rotate fencing token');
  assert.equal(second.attemptCount, 2, 'takeover should consume exactly one additional attempt');

  await assert.rejects(
    executeContentImportJob(auth(), {
      jobId: fixture.jobId,
      importType: 'questions',
      allowPartial: false,
      allowQueuedJob: true,
      leaseToken: first.leaseToken,
    }),
    error => error?.code === 'IMPORT_WORKER_LEASE_LOST',
    'old worker must be fenced before it can write imported content',
  );
  assert.equal(
    (await pool.query(`select count(*)::int as count from public.questions where tenant_id = $1 and legacy_id = $2`, [tenantId, fixture.legacyId])).rows[0].count,
    0,
    'fenced old worker must leave no business writes',
  );

  const staleFailure = await worker.markImportFailed(first, new Error('stale worker failure'));
  assert.equal(staleFailure, 'lease_lost', 'old worker must not schedule retry or failure after takeover');
  const afterStaleFailure = await readJob(pool, fixture.jobId);
  assert.equal(afterStaleFailure.leaseToken, second.leaseToken, 'stale failure must not overwrite current lease');

  const execution = await executeContentImportJob(auth(), {
    jobId: fixture.jobId,
    importType: 'questions',
    allowPartial: false,
    allowQueuedJob: true,
    leaseToken: second.leaseToken,
  });
  assert.equal(execution.status, 'completed', 'current lease owner should complete import');
  const completed = await readJob(pool, fixture.jobId);
  assert.equal(completed.status, 'completed');
  assert.equal(completed.attemptCount, 2, 'successful takeover must preserve exact attempt count');
  assert.equal(completed.leaseToken, null, 'terminal transition should release fencing token');
  assert.equal(Number(completed.insertedCount), 1);

  await assert.rejects(
    executeContentImportJob(auth(), {
      jobId: fixture.jobId,
      importType: 'questions',
      allowPartial: false,
      allowQueuedJob: true,
      leaseToken: first.leaseToken,
    }),
    error => error?.code === 'IMPORT_WORKER_LEASE_LOST',
    'old worker must not turn a completed takeover into an idempotent success',
  );
}

async function testRetryState(pool) {
  const fixture = await createQueuedQuestionImport(pool, 'retry');
  const [first] = await worker.claimImportJobs({ workerId: 'lease-retry-worker', batchSize: 1, leaseSeconds: 30 });
  const state = await worker.markImportFailed(first, Object.assign(new Error('retry fixture'), { code: 'RETRY_FIXTURE' }));
  assert.equal(state, 'retrying');
  const pending = await readJob(pool, fixture.jobId);
  assert.equal(pending.status, 'pending');
  assert.equal(pending.attemptCount, 1, 'retry scheduling must not increment attempt count');
  assert.equal(pending.leaseToken, null, 'retry scheduling must release lease');
  assert.ok(pending.nextAttemptAt, 'retry scheduling should persist next_attempt_at');

  await pool.query(
    `update public.content_import_jobs set next_attempt_at = now() - interval '1 second' where tenant_id = $1 and id = $2`,
    [tenantId, fixture.jobId],
  );
  const [second] = await worker.claimImportJobs({ workerId: 'lease-retry-worker-2', batchSize: 1, leaseSeconds: 30 });
  assert.equal(second.id, fixture.jobId);
  assert.equal(second.attemptCount, 2, 'retry claim should increment attempt exactly once');
  assert.notEqual(second.leaseToken, first.leaseToken, 'each attempt must receive a new fencing token');
}

async function testExpiredFinalAttempt(pool) {
  const fixture = await createQueuedQuestionImport(pool, 'exhausted', 1);
  await worker.claimImportJobs({ workerId: 'lease-crashed-final-worker', batchSize: 1, leaseSeconds: 30 });
  await pool.query(
    `
    update public.content_import_jobs
    set locked_at = now() - interval '10 seconds',
        lease_expires_at = now() - interval '1 second',
        last_heartbeat_at = now() - interval '10 seconds'
    where tenant_id = $1 and id = $2
    `,
    [tenantId, fixture.jobId],
  );
  const claimed = await worker.claimImportJobs({ workerId: 'lease-reaper-worker', batchSize: 1, leaseSeconds: 30 });
  assert.equal(claimed.length, 0, 'expired final attempt must not be executed again');
  const failed = await readJob(pool, fixture.jobId);
  assert.equal(failed.status, 'failed', 'expired final attempt should become terminal failed');
  assert.equal(failed.attemptCount, 1, 'reaping final attempt must not inflate attempts');
  assert.equal(failed.leaseToken, null, 'terminal reaper should release lease');
  assert.match(failed.errorMessage, /lease expired/i);
}

async function main() {
  const pool = new pg.Pool({ connectionString: databaseUrl, max: 12 });
  try {
    await assertDestructiveTestDatabase({
      client: pool,
      databaseUrl,
      confirmation,
      operation: 'import worker lease integration test',
    });
    await cleanup(pool);
    await testAtomicClaim(pool);
    await cleanup(pool);
    await testHeartbeat(pool);
    await cleanup(pool);
    await testExpiredTakeoverAndFencing(pool);
    await cleanup(pool);
    await testRetryState(pool);
    await cleanup(pool);
    await testExpiredFinalAttempt(pool);
    console.log('Import worker lease integration test complete.');
  } finally {
    await cleanup(pool).catch(() => undefined);
    await worker.closeImportExecutorPool().catch(() => undefined);
    await pool.end();
  }
}

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