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 ids = {
  tenant: '00000000-0000-0000-0000-00000000a899',
  authPlatformAdminUser: '00000000-0000-0000-0000-00000000a900',
  platformAdminUser: '00000000-0000-0000-0000-00000000a901',
  studentOne: '00000000-0000-0000-0000-00000000a902',
  studentTwo: '00000000-0000-0000-0000-00000000a903',
  disabledStudent: '00000000-0000-0000-0000-00000000a904',
  adminUser: '00000000-0000-0000-0000-00000000a905',
  questionOne: '00000000-0000-0000-0000-00000000a906',
  questionTwo: '00000000-0000-0000-0000-00000000a907',
  questionThree: '00000000-0000-0000-0000-00000000a908',
  questionDraft: '00000000-0000-0000-0000-00000000a909',
  assetOne: '00000000-0000-0000-0000-00000000a910',
  assetTwo: '00000000-0000-0000-0000-00000000a911',
  assetArchived: '00000000-0000-0000-0000-00000000a912',
  assetUnsafe: '00000000-0000-0000-0000-00000000a913',
  videoOne: '00000000-0000-0000-0000-00000000a914',
  videoTwo: '00000000-0000-0000-0000-00000000a915',
  videoInactive: '00000000-0000-0000-0000-00000000a916',
  orderOne: '00000000-0000-0000-0000-00000000a917',
  orderTwo: '00000000-0000-0000-0000-00000000a918',
  orderPending: '00000000-0000-0000-0000-00000000a919',
  orderOutside: '00000000-0000-0000-0000-00000000a920',
  entitlementOne: '00000000-0000-0000-0000-00000000a921',
  entitlementTwo: '00000000-0000-0000-0000-00000000a922',
  entitlementExpired: '00000000-0000-0000-0000-00000000a923',
  entitlementOutside: '00000000-0000-0000-0000-00000000a924',
  answerOne: '00000000-0000-0000-0000-00000000a925',
  answerTwo: '00000000-0000-0000-0000-00000000a926',
  answerOutside: '00000000-0000-0000-0000-00000000a927',
};

function runWorkerOnce() {
  const child = spawn(process.execPath, ['apps/worker/dist/apps/worker/src/index.js', '--once', '--job', 'platform-usage'], {
    cwd: process.cwd(),
    env: {
      ...process.env,
      DATABASE_URL: databaseUrl,
      WORKER_PLATFORM_USAGE_BATCH_SIZE: '1',
      WORKER_PLATFORM_USAGE_MONTH: '2026-06',
      WORKER_PLATFORM_USAGE_ID: 'platform-usage-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, /platform-usage batch processed=\d+/, 'worker output should include platform usage summary');
        resolve(output);
      } catch (error) {
        reject(error);
      }
    });
  });
}

async function cleanup(pool) {
  await pool.query('delete from public.tenants where id = $1', [ids.tenant]);
  await pool.query(
    `
    delete from public.platform_users
    where id = any($1::uuid[])
    `,
    [[ids.platformAdminUser, ids.studentOne, ids.studentTwo, ids.disabledStudent, ids.adminUser]],
  );
  await pool.query('delete from auth.users where id = $1', [ids.authPlatformAdminUser]).catch(() => {});
}

async function createAuthUser(pool) {
  await pool.query(
    `
    insert into auth.users (
      id, aud, role, phone, phone_confirmed_at, raw_app_meta_data, raw_user_meta_data, created_at, updated_at
    )
    values (
      $1, 'authenticated', 'authenticated', '13900000900', now(),
      '{"provider":"phone","providers":["phone"],"app_role":"platform_admin"}'::jsonb,
      '{}'::jsonb, now(), now()
    )
    on conflict (id)
    do update set phone = excluded.phone, updated_at = now()
    `,
    [ids.authPlatformAdminUser],
  ).catch(() => {});
  await pool.query(
    `
    insert into public.platform_users (
      id, auth_user_id, username, phone, name, primary_role, status, platform_permissions, raw_profile
    )
    values (
      $1, $2, 'usage_platform_admin', '13900000900', '用量平台管理员',
      'platform_admin', 'active', '{"*":true}'::jsonb, '{"source":"platform-usage-worker-test"}'::jsonb
    )
    `,
    [ids.platformAdminUser, ids.authPlatformAdminUser],
  );
}

async function createFixture(pool) {
  await createAuthUser(pool);

  await pool.query(
    `
    insert into public.tenants (
      id, slug, name, legal_name, status, mode, billing_status, metadata, created_at, updated_at
    )
    values (
      $1, 'platform-usage-worker', '平台用量采集测试租户', '平台用量采集测试有限公司',
      'active', 'saas', 'active', '{"source":"platform-usage-worker-test"}'::jsonb,
      '2000-01-01 00:00:00+00', now()
    )
    `,
    [ids.tenant],
  );

  await pool.query(
    `
    insert into public.platform_users (id, username, phone, name, primary_role, raw_profile)
    values
      ($1, 'usage_student_1', '13900000901', '用量学生一', 'student', '{"source":"platform-usage-worker-test"}'::jsonb),
      ($2, 'usage_student_2', '13900000902', '用量学生二', 'student', '{"source":"platform-usage-worker-test"}'::jsonb),
      ($3, 'usage_student_disabled', '13900000903', '禁用学生', 'student', '{"source":"platform-usage-worker-test"}'::jsonb),
      ($4, 'usage_admin', '13900000904', '用量管理员', 'tenant_admin', '{"source":"platform-usage-worker-test"}'::jsonb)
    `,
    [ids.studentOne, ids.studentTwo, ids.disabledStudent, ids.adminUser],
  );

  await pool.query(
    `
    insert into public.tenant_memberships (tenant_id, user_id, role, status, permissions)
    values
      ($1, $2, 'student', 'active', '{}'::jsonb),
      ($1, $3, 'student', 'active', '{}'::jsonb),
      ($1, $4, 'student', 'disabled', '{}'::jsonb),
      ($1, $5, 'tenant_admin', 'active', '{"platform:usage:read":false}'::jsonb)
    `,
    [ids.tenant, ids.studentOne, ids.studentTwo, ids.disabledStudent, ids.adminUser],
  );

  await pool.query(
    `
    insert into public.student_profiles (tenant_id, user_id, stats, progress)
    values
      ($1, $2, '{}'::jsonb, '{}'::jsonb),
      ($1, $3, '{}'::jsonb, '{}'::jsonb)
    `,
    [ids.tenant, ids.studentOne, ids.studentTwo],
  );

  await pool.query(
    `
    insert into public.questions (id, tenant_id, legacy_id, type, status)
    values
      ($1, $5, 'usage-q-1', 'choice', 'published'),
      ($2, $5, 'usage-q-2', 'choice', 'published'),
      ($3, $5, 'usage-q-3', 'choice', 'published'),
      ($4, $5, 'usage-q-draft', 'choice', 'draft')
    `,
    [ids.questionOne, ids.questionTwo, ids.questionThree, ids.questionDraft, ids.tenant],
  );

  await pool.query(
    `
    insert into public.answer_records (
      id, tenant_id, user_id, question_id, legacy_id, selected_options, is_correct, answered_at
    )
    values
      ($1, $4, $5, $7, 'usage-answer-1', '[0]'::jsonb, true, '2026-06-10 08:00:00+00'),
      ($2, $4, $6, $8, 'usage-answer-2', '[1]'::jsonb, false, '2026-06-15 08:00:00+00'),
      ($3, $4, $5, $9, 'usage-answer-outside', '[0]'::jsonb, true, '2026-05-15 08:00:00+00')
    `,
    [
      ids.answerOne,
      ids.answerTwo,
      ids.answerOutside,
      ids.tenant,
      ids.studentOne,
      ids.studentTwo,
      ids.questionOne,
      ids.questionTwo,
      ids.questionThree,
    ],
  );

  await pool.query(
    `
    insert into public.content_assets (
      id, tenant_id, legacy_id, asset_key, title, asset_type, storage_provider,
      bucket, object_key, file_name, mime_type, file_size_bytes,
      visibility, status, upload_status, verified_size_bytes,
      security_scan_status, security_scan_summary, source
    )
    values
      ($1, $5, 'usage-asset-1', 'usage/asset-1.pdf', '已验证资料', 'pdf', 'local_dev',
       'tenant-assets', 'usage/asset-1.pdf', 'asset-1.pdf', 'application/pdf', 1073741824,
       'tenant', 'active', 'verified', 1073741824, 'passed', '{"riskLevel":"none"}'::jsonb, 'test'),
      ($2, $5, 'usage-asset-2', 'usage/asset-2.pdf', '外链资料', 'pdf', 'external_url',
       null, null, 'asset-2.pdf', 'application/pdf', 536870912,
       'tenant', 'active', 'not_required', null, 'not_required', '{}'::jsonb, 'test'),
      ($3, $5, 'usage-asset-archived', 'usage/asset-archived.pdf', '归档资料', 'pdf', 'local_dev',
       'tenant-assets', 'usage/asset-archived.pdf', 'asset-archived.pdf', 'application/pdf', 5368709120,
       'tenant', 'archived', 'verified', 5368709120, 'passed', '{"riskLevel":"none"}'::jsonb, 'test'),
      ($4, $5, 'usage-asset-unsafe', 'usage/asset-unsafe.pdf', '未通过扫描资料', 'pdf', 'local_dev',
       'tenant-assets', 'usage/asset-unsafe.pdf', 'asset-unsafe.pdf', 'application/pdf', 2147483648,
       'tenant', 'active', 'verified', 2147483648, 'failed', '{"riskLevel":"high"}'::jsonb, 'test')
    `,
    [ids.assetOne, ids.assetTwo, ids.assetArchived, ids.assetUnsafe, ids.tenant],
  );

  await pool.query(
    `
    insert into public.video_explanations (
      id, tenant_id, legacy_id, title, video_url, duration_seconds, is_active, access_mode, play_count
    )
    values
      ($1, $4, 'usage-video-1', '用量视频一', 'https://example.test/video-1.mp4', 120, true, 'svip', 0),
      ($2, $4, 'usage-video-2', '用量视频二', 'https://example.test/video-2.mp4', 180, true, 'video_quota', 0),
      ($3, $4, 'usage-video-inactive', '停用视频', 'https://example.test/video-3.mp4', 90, false, 'svip', 0)
    `,
    [ids.videoOne, ids.videoTwo, ids.videoInactive, ids.tenant],
  );

  await pool.query(
    `
    insert into public.video_play_events (
      tenant_id, user_id, video_id, question_id, play_token_hash,
      status, access_mode, consumed_quota, signed_url_expires_at, metadata, created_at, updated_at
    )
    values
      ($1, $2, $4, $6, 'usage-play-1', 'issued', 'svip', 0, '2026-06-10 09:00:00+00', '{}'::jsonb, '2026-06-10 08:00:00+00', '2026-06-10 08:00:00+00'),
      ($1, $3, $5, $7, 'usage-play-2', 'completed', 'video_quota', 1, '2026-06-12 09:00:00+00', '{}'::jsonb, '2026-06-12 08:00:00+00', '2026-06-12 08:00:00+00'),
      ($1, $2, $5, $7, 'usage-play-3', 'started', 'video_quota', 2, '2026-06-20 09:00:00+00', '{}'::jsonb, '2026-06-20 08:00:00+00', '2026-06-20 08:00:00+00'),
      ($1, $2, $4, $6, 'usage-play-outside', 'issued', 'svip', 1, '2026-05-10 09:00:00+00', '{}'::jsonb, '2026-05-10 08:00:00+00', '2026-05-10 08:00:00+00')
    `,
    [
      ids.tenant,
      ids.studentOne,
      ids.studentTwo,
      ids.videoOne,
      ids.videoTwo,
      ids.questionOne,
      ids.questionTwo,
    ],
  );

  await pool.query(
    `
    insert into public.orders (
      id, tenant_id, user_id, legacy_id, order_no, status, product_type,
      product_name, amount_cents, pay_method, pay_provider, paid_at, created_at, updated_at
    )
    values
      ($1, $5, $6, 'usage-order-1', 'USAGE202606001', 'paid', 'svip', '用量套餐一', 1000, 'wechat', 'wechat_pay', '2026-06-05 08:00:00+00', '2026-06-05 07:00:00+00', '2026-06-05 08:00:00+00'),
      ($2, $5, $7, 'usage-order-2', 'USAGE202606002', 'paid', 'svip', '用量套餐二', 2500, 'alipay', 'alipay', '2026-06-18 08:00:00+00', '2026-06-18 07:00:00+00', '2026-06-18 08:00:00+00'),
      ($3, $5, $6, 'usage-order-pending', 'USAGE202606003', 'pending', 'svip', '待支付套餐', 9000, null, null, null, '2026-06-20 07:00:00+00', '2026-06-20 07:00:00+00'),
      ($4, $5, $6, 'usage-order-outside', 'USAGE202607001', 'paid', 'svip', '七月套餐', 7000, 'wechat', 'wechat_pay', '2026-07-01 08:00:00+00', '2026-07-01 07:00:00+00', '2026-07-01 08:00:00+00')
    `,
    [ids.orderOne, ids.orderTwo, ids.orderPending, ids.orderOutside, ids.tenant, ids.studentOne, ids.studentTwo],
  );

  await pool.query(
    `
    insert into public.entitlements (
      id, tenant_id, user_id, entitlement_type, scope_type,
      source_type, source_id, starts_at, expires_at, status, metadata
    )
    values
      ($1, $5, $6, 'svip', 'tenant', 'order', $8, '2026-06-01 00:00:00+00', null, 'active', '{}'::jsonb),
      ($2, $5, $7, 'svip', 'tenant', 'order', $9, '2026-05-01 00:00:00+00', '2026-06-10 00:00:00+00', 'active', '{}'::jsonb),
      ($3, $5, $6, 'svip', 'tenant', 'order', $8, '2026-05-01 00:00:00+00', '2026-06-10 00:00:00+00', 'expired', '{}'::jsonb),
      ($4, $5, $6, 'svip', 'tenant', 'order', $10, '2026-07-01 00:00:00+00', null, 'active', '{}'::jsonb)
    `,
    [
      ids.entitlementOne,
      ids.entitlementTwo,
      ids.entitlementExpired,
      ids.entitlementOutside,
      ids.tenant,
      ids.studentOne,
      ids.studentTwo,
      ids.orderOne,
      ids.orderTwo,
      ids.orderOutside,
    ],
  );

  await pool.query(
    `
    insert into public.tenant_usage_records (
      tenant_id, metric_key, metric_value, period_start, period_end, metadata
    )
    values ($1, 'students', 999, '2026-06-01', '2026-06-30', '{"source":"manual_adjustment","note":"must not be overwritten"}'::jsonb)
    `,
    [ids.tenant],
  );
}

async function usageMap(pool) {
  const records = await pool.query(
    `
    select metric_key as "metricKey", metric_value as "metricValue", metadata
    from public.tenant_usage_records
    where tenant_id = $1
      and period_start = '2026-06-01'
      and period_end = '2026-06-30'
      and metadata->>'source' = 'platform_usage_worker'
    order by metric_key asc
    `,
    [ids.tenant],
  );
  return new Map(records.rows.map(row => [row.metricKey, row]));
}

function assertMetric(metrics, key, expected) {
  assert.ok(metrics.has(key), `metric ${key} should exist`);
  assert.equal(Number(metrics.get(key).metricValue), expected, `metric ${key} should match`);
}

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

    const firstOutput = await runWorkerOnce();
    assert.match(firstOutput, /processed=\d+/, 'worker should process active tenants');
    assert.match(firstOutput, /created=\d+/, 'first worker run should create usage metrics');
    assert.match(firstOutput, /failed=0/, 'first worker run should not fail');

    const firstMetrics = await usageMap(pool);
    assert.equal(firstMetrics.size, 11, 'worker should create one record per supported metric');
    assertMetric(firstMetrics, 'students', 2);
    assertMetric(firstMetrics, 'active_students', 2);
    assertMetric(firstMetrics, 'questions', 3);
    assertMetric(firstMetrics, 'assets', 3);
    assertMetric(firstMetrics, 'videos', 2);
    assertMetric(firstMetrics, 'video_plays', 3);
    assertMetric(firstMetrics, 'video_quota_consumed', 3);
    assertMetric(firstMetrics, 'paid_orders', 2);
    assertMetric(firstMetrics, 'paid_order_amount_cents', 3500);
    assertMetric(firstMetrics, 'active_entitlements', 2);
    assert.ok(Math.abs(Number(firstMetrics.get('storage_gb').metricValue) - 1.5) < 0.000001, 'storage_gb should use verified safe asset size');
    assert.equal(firstMetrics.get('storage_gb').metadata?.bytes, 1610612736, 'storage metric should keep bytes in metadata');
    assert.equal(firstMetrics.get('students').metadata?.workerId, 'platform-usage-integration-test', 'metadata should record worker id');

    const manualRecords = await pool.query(
      `
      select metric_value as "metricValue", metadata
      from public.tenant_usage_records
      where tenant_id = $1
        and metric_key = 'students'
        and metadata->>'source' = 'manual_adjustment'
      `,
      [ids.tenant],
    );
    assert.equal(manualRecords.rowCount, 1, 'manual usage record should remain append-only');
    assert.equal(Number(manualRecords.rows[0]?.metricValue), 999, 'manual usage value should not be overwritten by worker');

    const secondOutput = await runWorkerOnce();
    assert.match(secondOutput, /created=0/, 'second worker run should not duplicate usage metrics');
    assert.match(secondOutput, /updated=\d+/, 'second worker run should update existing usage metrics');

    const workerRecordCount = await pool.query(
      `
      select count(*)::integer as count
      from public.tenant_usage_records
      where tenant_id = $1
        and period_start = '2026-06-01'
        and period_end = '2026-06-30'
        and metadata->>'source' = 'platform_usage_worker'
      `,
      [ids.tenant],
    );
    assert.equal(workerRecordCount.rows[0]?.count, 11, 'worker should keep exactly one record per metric after rerun');

    const detailUsageResult = await pool.query(
      `
      select metric_key as "metricKey", metric_value as "metricValue", metadata
      from public.tenant_usage_records
      where tenant_id = $1
        and metric_key = 'students'
        and period_start = '2026-06-01'
        and period_end = '2026-06-30'
      order by created_at desc
      `,
      [ids.tenant],
    );
    const detailUsage = detailUsageResult.rows;
    assert.ok(
      detailUsage.some(item => item.metricKey === 'students' && item.metadata?.source === 'platform_usage_worker' && Number(item.metricValue) === 2),
      'usage records should include generated worker usage',
    );
    assert.ok(
      detailUsage.some(item => item.metricKey === 'students' && item.metadata?.source === 'manual_adjustment' && Number(item.metricValue) === 999),
      'usage records should keep manual usage adjustment records for audit',
    );

    const audit = await pool.query(
      `
      select action, details
      from public.audit_logs
      where tenant_id = $1
        and action = 'platform.usage.worker_collected'
      order by created_at desc
      limit 1
      `,
      [ids.tenant],
    );
    assert.equal(audit.rows[0]?.details?.metricKeys?.length, 11, 'worker should audit collected metric keys');

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

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