// Isolated PostgreSQL and Redis verification. Never calls a Gateway or business queue.
import assert from 'node:assert/strict';
import { randomUUID } from 'node:crypto';
import { readFileSync } from 'node:fs';
import pg from '../../api/node_modules/pg/lib/index.js';
import prisma from '../../api/node_modules/@prisma/client/default.js';
import adapter from '../../api/node_modules/@prisma/adapter-pg/dist/index.js';
import night from '../../api/dist/risk-review/night-sending-risk.service.js';
import risk from '../../api/dist/risk-review/risk-review.service.js';
import continuation from '../../api/dist/send-chain/send-review-continuation.service.js';
import bullmq from '../../api/node_modules/bullmq/dist/cjs/index.js';

const connectionString = process.env.QA_DATABASE_URL || process.env.DATABASE_URL;
if (!connectionString) throw Error('QA_DATABASE_URL is required');
const schema = `qa_night_${randomUUID().replaceAll('-', '')}`;
assert.match(schema, /^qa_night_[a-f0-9]{32}$/);
const admin = new pg.Client({ connectionString });
const clients = [],
  checks = [];
let queue;
await admin.connect();
try {
  const before = (
    await admin.query(
      'SELECT (SELECT count(*) FROM public."SmsMessageRecord") messages,(SELECT count(*) FROM public."SmsSubmitRecord") submits',
    )
  ).rows[0];
  await admin.query(`CREATE SCHEMA "${schema}"`);
  for (const table of [
    'Tenant',
    'SmsApplication',
    'SmsBatchTask',
    'SmsMessageRecord',
    'SmsSubmitRecord',
    'SmsSendTask',
    'RiskRule',
    'RiskHitRecord',
    'User',
    'SmsTemplate',
    'TemplateVariable',
    'SmsSignature',
    'SensitiveWord',
  ]) {
    await admin.query(`CREATE TABLE "${schema}"."${table}" (LIKE public."${table}" INCLUDING ALL)`);
  }
  await admin.query(`SET search_path TO "${schema}"`);
  // LIKE also copies these columns after the release; recreate them only inside
  // this isolated schema so the forward migration is tested on every run.
  await admin.query(
    'ALTER TABLE "SmsSendTask" DROP COLUMN IF EXISTS "continuationLeaseOwner", DROP COLUMN IF EXISTS "continuationLeaseExpiresAt"',
  );
  await admin.query(
    readFileSync(
      new URL('../../api/prisma/migrations/20260907143000_night_sending_risk/migration.sql', import.meta.url),
      'utf8',
    ),
  );
  for (let i = 0; i < 2; i++) {
    const pool = new pg.Pool({ connectionString, max: 4, options: `-c search_path=${schema} -c timezone=UTC` });
    clients.push(
      new prisma.PrismaClient({ adapter: new adapter.PrismaPg(pool, { schema, disposeExternalPool: true }) }),
    );
  }
  const [db, db2] = clients;
  assert.equal((await db.$queryRawUnsafe('SELECT current_schema() AS name'))[0].name, schema);
  for (const tenantId of ['t1', 't2'])
    await db.tenant.create({ data: { id: tenantId, code: tenantId, name: tenantId } });
  for (const [id, tenantId] of [
    ['a1', 't1'],
    ['a2', 't1'],
    ['a3', 't1'],
    ['b1', 't2'],
  ])
    await db.smsApplication.create({
      data: { id, tenantId, name: id, cmppAccount: id, cmppEnterpriseCode: 'qa', secretHash: 'isolated-unused' },
    });
  const ruleData = {
    code: night.NIGHT_RULE_CODE,
    name: '夜间累计发送量审核',
    metric: 'nightSendingCount',
    thresholdValue: 5000,
    action: 'manual_review',
    status: 'active',
    config: { startTime: '21:00', endTime: '08:00', timeZone: 'Asia/Shanghai' },
  };
  const global = await db.riskRule.create({ data: { ...ruleData, id: 'global' } });
  const now = new Date('2026-09-06T14:00:01Z');
  const window = night.nightWindow(now, night.nightClock(null));
  await db.nightSendingWindow.create({
    data: {
      id: `a1:${window.windowStartedAt.toISOString()}`,
      tenantId: 't1',
      applicationId: 'a1',
      ...window,
      count: 4990,
      baselineCount: 4990,
    },
  });
  let sequence = 0;
  async function message(app = 'a1', content = '【测试】通知验证码和任意内容', sourceType = 'cmpp', extra = {}) {
    const id = `m${++sequence}`,
      tenantId = app === 'b1' ? 't2' : 't1';
    await db.smsBatchTask.create({
      data: {
        id: `task-${id}`,
        taskNo: `task-${id}`,
        tenantId,
        applicationId: app,
        content,
        phoneTotal: 1,
        sourceType,
        status: 'queued',
      },
    });
    return db.smsMessageRecord.create({
      data: {
        id,
        messageId: id,
        tenantId,
        applicationId: app,
        batchTaskId: `task-${id}`,
        phoneNumber: '13800000001',
        content,
        status: 'queued',
        ...extra,
      },
    });
  }
  const services = [new night.NightSendingRiskService(db), new night.NightSendingRiskService(db2)];
  const messages = [];
  for (let i = 0; i < 20; i++) messages.push(await message('a1', undefined, ['cmpp', 'api', 'client'][i % 3]));
  const results = await Promise.all(messages.map((m, i) => services[i % 2].guard([m.id], now)));
  assert.equal(
    results.reduce((sum, result) => sum + result.size, 0),
    10,
  );
  assert.equal((await db.nightSendingWindow.findFirst({ where: { applicationId: 'a1' } })).count, 5010);
  assert.equal(await db.nightSendingReservation.count(), 20);
  const held = await db.smsMessageRecord.findMany({ where: { status: 'pending_review' } });
  const tasks = await db.smsSendTask.findMany();
  assert.equal(tasks.length, 1);
  assert.equal(tasks[0].phoneTotal, 10);
  assert.equal(tasks[0].uniquePhoneTotal, 1);
  checks.push('mixed sources + 2 instances concurrent boundary 4990->5010; same content aggregation');
  await Promise.all(messages.map((m, i) => services[i % 2].guard([m.id], now)));
  assert.equal((await db.nightSendingWindow.findFirst({ where: { applicationId: 'a1' } })).count, 5010);
  checks.push('duplicate jobs are idempotent');
  const other = await message('a1', '另一种内容');
  await services[0].guard([other.id], now);
  assert.equal(await db.smsSendTask.count(), 2);
  const midnight = await message();
  await services[1].guard([midnight.id], new Date('2026-09-06T16:00:00Z'));
  assert.equal((await db.nightSendingWindow.findFirst({ where: { applicationId: 'a1' } })).count, 5012);
  checks.push('different content separate review; midnight retains application total');
  const a2 = await message('a2'),
    b1 = await message('b1');
  assert.equal((await services[0].guard([a2.id, b1.id], now)).size, 0);
  assert.equal((await db.nightSendingWindow.findFirst({ where: { applicationId: 'a2' } })).count, 1);
  checks.push('application and tenant isolation');
  const override = await db.riskRule.create({
    data: { ...ruleData, applicationId: 'a2', tenantId: 't1', thresholdValue: 1 },
  });
  const overrideMessage = await message('a2');
  assert.equal((await services[0].guard([overrideMessage.id], now)).size, 1);
  await db.riskRule.update({ where: { id: override.id }, data: { status: 'inactive' } });
  const fallback = await message('a2');
  assert.equal((await services[0].guard([fallback.id], now)).size, 0);
  assert.equal((await db.nightSendingWindow.findFirst({ where: { applicationId: 'a2' } })).count, 3);
  checks.push('individual threshold override + disable restores global without clearing count');
  const day = await message('b1');
  assert.equal((await services[0].guard([day.id], new Date('2026-09-07T04:00:00Z'))).size, 0);
  assert.equal(await db.nightSendingReservation.count({ where: { messageRecordId: day.id } }), 0);
  await services[0].guard([day.id], new Date('2026-09-07T13:00:00Z'));
  assert.equal(await db.nightSendingReservation.count({ where: { messageRecordId: day.id } }), 1);
  checks.push('daytime queue + nighttime execution/scheduled task evaluated at dispatch');
  const retry = await message('a1', undefined, 'cmpp', { submitId: 'already-submitted', billingUnits: 3 });
  await services[0].guard([retry.id], now);
  assert.equal(await db.nightSendingReservation.count({ where: { messageRecordId: retry.id } }), 0);
  checks.push('retry/segments do not consume new allowance');
  const legacy = await message('a3', '旧版本本夜首次提交', 'api', { status: 'submitted', submitId: 'legacy' });
  const previousDay = await message('a3', '昨日首次提交本夜重试', 'cmpp', {
    status: 'submitted',
    submitId: 'previous',
  });
  for (const [id, messageRecordId, createdAt] of [
    ['legacy-first', legacy.id, now],
    ['legacy-retry', legacy.id, now],
    ['previous-first', previousDay.id, new Date(now.getTime() - 86_400_000)],
    ['previous-retry', previousDay.id, now],
  ])
    await db.smsSubmitRecord.create({
      data: { id, submitId: id, messageRecordId, channelId: 'isolated-unused', createdAt },
    });
  const newAfterDeploy = await message('a3');
  await services[0].guard([newAfterDeploy.id], now);
  const initialized = await db.nightSendingWindow.findFirst({ where: { applicationId: 'a3' } });
  assert.equal(initialized.baselineCount, 1);
  assert.equal(initialized.count, 2);
  checks.push('midnight deployment bootstrap counts first business submissions, not retry attempts');
  const review = new risk.RiskReviewService(db);
  const changed = await review.updateRule(global.id, {
    thresholdValue: 1,
    config: { startTime: '23:00', endTime: '07:00', timeZone: 'Asia/Shanghai' },
  });
  assert.equal(changed.thresholdValue, 1);
  await assert.rejects(review.updateRule(global.id, { thresholdValue: 1.5 }));
  await assert.rejects(review.updateRule(global.id, { action: 'block' }));
  await assert.rejects(
    review.updateRule(global.id, { config: { startTime: '21:00', endTime: '08:00', timeZone: 'UTC' } }),
  );
  checks.push('rule write validation and forced manual review');
  await review.approveTask(tasks[0].id, { reason: '隔离审核' });
  await review.approveTask(tasks[0].id, { reason: '重复请求' });
  await assert.rejects(review.rejectTask(tasks[0].id, { reason: '相反决定' }));
  const redis = new URL(process.env.REDIS_URL || 'redis://127.0.0.1:6379');
  queue = new bullmq.Queue(schema, {
    connection: { host: redis.hostname, port: Number(redis.port || 6379), password: redis.password || undefined },
  });
  let failQueue = true;
  const facade = {
    getSendQueue: () => ({
      add: (...args) => {
        if (failQueue) throw Error('isolated injected queue outage');
        return queue.add(...args);
      },
    }),
    refreshTaskProgress: async () => {},
  };
  const resumes = new continuation.SendReviewContinuationService(db, {}, {}, {}, {}, facade, {
    releaseMessageReservation: async () => {},
    recordCmppFailureReceipt: async () => {},
  });
  await assert.rejects(resumes.handleReviewDecision(tasks[0].id, 'approved', '隔离审核'));
  assert((await review.pendingNightContinuations()).some((t) => t.id === tasks[0].id));
  failQueue = false;
  await Promise.all([
    resumes.handleReviewDecision(tasks[0].id, 'approved', '隔离审核'),
    resumes.handleReviewDecision(tasks[0].id, 'approved', '隔离审核'),
  ]);
  const queued = await queue.getJobCounts('wait', 'prioritized');
  assert.equal(queued.wait + queued.prioritized, 10);
  assert.equal(await db.nightSendingReservation.count({ where: { reviewTaskId: tasks[0].id, continuedAt: null } }), 0);
  for (const m of held) assert.equal((await services[0].guard([m.id], now)).size, 0);
  assert.equal((await db.smsMessageRecord.findUnique({ where: { id: other.id } })).status, 'pending_review');
  checks.push('review idempotency/conflict + real Redis recovery + approved range only');
  const invalid = await message('a2');
  await admin.query(
    `ALTER TABLE "${schema}"."NightSendingReservation" ADD CONSTRAINT injected_failure CHECK ("messageRecordId" <> '${invalid.id}')`,
  );
  const countBefore = (await db.nightSendingWindow.findFirst({ where: { applicationId: 'a2' } })).count;
  await assert.rejects(services[0].guard([invalid.id], now));
  assert.equal((await db.nightSendingWindow.findFirst({ where: { applicationId: 'a2' } })).count, countBefore);
  assert.equal((await db.smsMessageRecord.findUnique({ where: { id: invalid.id } })).status, 'queued');
  checks.push('transaction failure rolls back counter and review atomically');
  const after = (
    await admin.query(
      'SELECT (SELECT count(*) FROM public."SmsMessageRecord") messages,(SELECT count(*) FROM public."SmsSubmitRecord") submits',
    )
  ).rows[0];
  assert.deepEqual(after, before);
  console.log(
    JSON.stringify(
      {
        passed: checks.length,
        checks,
        schema,
        publicBefore: before,
        publicAfter: after,
        businessQueueWrites: 0,
        gatewayCalls: 0,
      },
      null,
      2,
    ),
  );
} finally {
  if (queue) {
    await queue.obliterate({ force: true });
    await queue.close();
  }
  for (const client of clients) await client.$disconnect();
  await admin.query(`DROP SCHEMA IF EXISTS "${schema}" CASCADE`);
  await admin.end();
}