285 lines
12 KiB
JavaScript
285 lines
12 KiB
JavaScript
// 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();
|
|
}
|