253 lines
12 KiB
JavaScript
253 lines
12 KiB
JavaScript
import pg from '../../api/node_modules/pg/lib/index.js';
|
|
import { readFileSync } from 'node:fs';
|
|
import { resolve } from 'node:path';
|
|
import assert from 'node:assert/strict';
|
|
import { randomUUID } from 'node:crypto';
|
|
import projection from '../../api/dist/sending-monitor/monitor-projection.js';
|
|
import evaluation from '../../api/dist/sending-monitor/monitor-evaluation.js';
|
|
const { Client } = pg;
|
|
const { projectMessages, keyOf } = projection;
|
|
const { evaluateWindow } = evaluation;
|
|
process.env.TZ = 'UTC';
|
|
async function main() {
|
|
const connectionString = process.env.QA_DATABASE_URL || process.env.DATABASE_URL;
|
|
if (!connectionString) throw new Error('QA_DATABASE_URL is required');
|
|
const db = new Client({ connectionString, application_name: 'cmpp-monitor-isolated-qa' });
|
|
const schema = `qa_monitor_${randomUUID().replaceAll('-', '')}`;
|
|
if (!/^qa_monitor_[a-f0-9]{32}$/.test(schema)) throw new Error('Invalid isolated schema');
|
|
let checks = 0;
|
|
await db.connect();
|
|
try {
|
|
await db.query('BEGIN');
|
|
await db.query(`CREATE SCHEMA "${schema}"`);
|
|
await db.query(`SET LOCAL search_path TO "${schema}",public`);
|
|
await db.query(`SET LOCAL timezone='Asia/Shanghai'`);
|
|
for (const table of [
|
|
'Tenant',
|
|
'SmsApplication',
|
|
'SmsSignature',
|
|
'SmsDrainageInfo',
|
|
'SmsChannel',
|
|
'ChannelSignatureReportTask',
|
|
'SmsMessageRecord',
|
|
'SmsSubmitRecord',
|
|
'SmsMessageSegmentAudit',
|
|
'UpstreamReceiptInbox',
|
|
]) {
|
|
await db.query(`CREATE TABLE "${table}" (LIKE public."${table}" INCLUDING ALL)`);
|
|
if (['Tenant', 'SmsApplication', 'SmsSignature', 'SmsDrainageInfo', 'SmsChannel'].includes(table))
|
|
await db.query(`INSERT INTO "${table}" SELECT * FROM public."${table}"`);
|
|
}
|
|
for (const table of ['SmsSubmitRecord', 'SmsMessageSegmentAudit'])
|
|
await db.query(
|
|
`ALTER TABLE "${table}" DROP COLUMN IF EXISTS "firstWireSubmitAt", DROP COLUMN IF EXISTS "wireTimeSource", DROP COLUMN IF EXISTS "receiptRequested"`,
|
|
);
|
|
await db.query(`ALTER TABLE "UpstreamReceiptInbox" DROP COLUMN IF EXISTS "gatewayReceivedAt"`);
|
|
// LIKE copies indexes under generated names; explicit migration index names are unique within the new schema.
|
|
for (const path of ['20260906170000_report_readiness_notifications', '20260906171000_sending_monitor'])
|
|
await db.query(
|
|
readFileSync(resolve(import.meta.dirname, '../../api/prisma/migrations', path, 'migration.sql'), 'utf8'),
|
|
);
|
|
const sig = (
|
|
await db.query(
|
|
`SELECT id,"tenantId","applicationId" FROM "SmsSignature" WHERE "applicationId" IS NOT NULL LIMIT 1`,
|
|
)
|
|
).rows[0];
|
|
const channel = (await db.query(`SELECT id FROM "SmsChannel" LIMIT 1`)).rows[0].id;
|
|
assert(sig && channel, 'Need existing metadata to copy into isolated fixtures');
|
|
await db.query(
|
|
`UPDATE "SmsChannel" SET status='active',carrier='all',carriers=ARRAY['mobile','unicom','telecom'],"sendRegion"='全国' WHERE id=$1`,
|
|
[channel],
|
|
);
|
|
for (const carrier of ['mobile', 'unicom', 'telecom']) {
|
|
await db.query(
|
|
`INSERT INTO "ChannelSignatureReportTask" (id,"tenantId","signatureId","channelId",carrier,"approvalScope",status,"updatedAt") VALUES($1,$2,$3,$4,$5,'carrier_specific','approved',CURRENT_TIMESTAMP)`,
|
|
[`qa-${carrier}`, sig.tenantId, sig.id, channel, carrier],
|
|
);
|
|
assert.equal(
|
|
Number((await db.query(`SELECT count(*) n FROM "ReportReadinessEvent"`)).rows[0].n),
|
|
carrier === 'telecom' ? 1 : 0,
|
|
);
|
|
checks++;
|
|
}
|
|
await db.query(`UPDATE "ChannelSignatureReportTask" SET status='approved'`);
|
|
assert.equal(Number((await db.query(`SELECT count(*) n FROM "ReportReadinessEvent"`)).rows[0].n), 1);
|
|
checks++;
|
|
await db.query(`UPDATE "ChannelSignatureReportTask" SET status='failed'`);
|
|
await db.query(`UPDATE "ChannelSignatureReportTask" SET status='approved'`);
|
|
assert.equal(Number((await db.query(`SELECT revision FROM "ReportNotificationHour"`)).rows[0].revision), 2);
|
|
checks++;
|
|
await db.query(`UPDATE "ChannelSignatureReportTask" SET status='failed' WHERE carrier='mobile'`);
|
|
await db.query(`UPDATE "ChannelSignatureReportTask" SET status='approved' WHERE carrier='mobile'`);
|
|
assert.equal(Number((await db.query(`SELECT count(*) n FROM "ReportReadinessEvent"`)).rows[0].n), 2);
|
|
checks++;
|
|
await db.query(`UPDATE "SmsChannel" SET "sendRegion"='广东' WHERE id=$1`, [channel]);
|
|
assert.equal((await db.query(`SELECT cmpp_report_ready_mask('signature',$1,NULL) mask`, [sig.id])).rows[0].mask, 0);
|
|
checks++;
|
|
await db.query(`UPDATE "SmsChannel" SET "sendRegion"='全国' WHERE id=$1`, [channel]);
|
|
|
|
const t = new Date(Math.floor(Date.now() / 300000) * 300000);
|
|
const earlier = new Date(t.getTime() - 60000),
|
|
late = new Date(t.getTime() - 1000);
|
|
await db.query(
|
|
`INSERT INTO "SendingMonitorTarget" ("channelId",enabled,version,"effectiveFrom","updatedBy") VALUES($1,true,1,$2,'qa')`,
|
|
[channel, new Date(t.getTime() - 3600000)],
|
|
);
|
|
await db.query(
|
|
`INSERT INTO "SendingMonitorTargetVersion" SELECT "channelId",version,enabled,"effectiveFrom","updatedBy" FROM "SendingMonitorTarget"`,
|
|
);
|
|
const config = { enabled: true, minSamples: 1, thresholds: [90, 95, 98], consecutiveBad: 1, consecutiveGood: 2 };
|
|
await db.query(
|
|
`INSERT INTO "SendingMonitorRuleVersion" ("ruleId",version,type,scope,config,"effectiveAt","createdBy") VALUES('qa-rule',1,'industry','{}',$1::jsonb,$2,'qa')`,
|
|
[JSON.stringify(config), new Date(t.getTime() - 3600000)],
|
|
);
|
|
// These are isolated, never-enqueued records, explicitly synthetic, stored and queried by actual PostgreSQL.
|
|
for (let n = 0; n < 4; n++) {
|
|
const id = `qa-message-${n}`,
|
|
submitId = `qa-submit-${n}`,
|
|
at = n === 3 ? late : earlier;
|
|
await db.query(
|
|
`INSERT INTO "SmsMessageRecord" (id,"tenantId","applicationId","signatureId","messageId","phoneNumber",carrier,content,"cmppRegisteredDelivery","updatedAt") VALUES($1,$2,$3,$4,$1,'13800000000','mobile','【隔离验收】验证码',true,CURRENT_TIMESTAMP)`,
|
|
[id, sig.tenantId, sig.applicationId, sig.id],
|
|
);
|
|
await db.query(
|
|
`INSERT INTO "SmsSubmitRecord" (id,"messageRecordId","channelId","submitId","firstWireSubmitAt","wireTimeSource","receiptRequested","createdAt","updatedAt") VALUES($1,$2,$3,$1,$4,'gateway_write_complete',true,$4,CURRENT_TIMESTAMP)`,
|
|
[submitId, id, channel, at],
|
|
);
|
|
await db.query(
|
|
`INSERT INTO "SmsMessageSegmentAudit" (id,"messageRecordId","submitRecordId","channelId","submitId","gatewayMessageId","segmentTotal","segmentIndex","firstWireSubmitAt","wireTimeSource","updatedAt") VALUES($1,$2,$3,$4,$3,$1,1,1,$5,'gateway_write_complete',CURRENT_TIMESTAMP)`,
|
|
[`qa-gateway-${n}`, id, submitId, channel, at],
|
|
);
|
|
const receiptAt = new Date(at.getTime() + [4999, 5000, 5001, 50000][n]);
|
|
await db.query(
|
|
`INSERT INTO "UpstreamReceiptInbox" (id,"receiptKey","incomingChannelId","upstreamAccount","upstreamHost","upstreamPort",protocol,"protocolVersion","gatewayMessageId","receiptStatus","rawStatus","deliveredAt","gatewayReceivedAt",status,"matchedMessageRecordId","matchedChannelId","updatedAt") VALUES($1,$1,$2,'qa','127.0.0.1',7890,'cmpp','3.0',$3,'delivered','DELIVRD',$4,$4,'matched',$5,$2,CURRENT_TIMESTAMP)`,
|
|
[`qa-receipt-${n}`, channel, `qa-gateway-${n}`, receiptAt, id],
|
|
);
|
|
}
|
|
const ids = [0, 1, 2, 3].map((n) => `qa-message-${n}`);
|
|
await projectMessages(db, ids);
|
|
const facts = (await db.query(`SELECT count(*)::int count FROM "SendingMonitorFact"`)).rows[0].count;
|
|
assert.equal(facts, 8);
|
|
checks++;
|
|
await projectMessages(db, ids);
|
|
assert.equal((await db.query(`SELECT count(*)::int count FROM "SendingMonitorFact"`)).rows[0].count, facts);
|
|
checks++;
|
|
await evaluateWindow(db, 'industry', t, false, true);
|
|
const key = keyOf([channel, 'mobile']);
|
|
let row = (await db.query(`SELECT * FROM "SendingMonitorSnapshot" WHERE "dimensionKey"=$1`, [key])).rows[0];
|
|
assert.equal(row.metrics.total, 4);
|
|
assert.deepEqual(
|
|
row.metrics.metrics.map((m) => [m.success, m.mature, m.observing]),
|
|
[
|
|
[2, 3, 1],
|
|
[3, 3, 1],
|
|
[3, 3, 1],
|
|
],
|
|
);
|
|
checks++;
|
|
await evaluateWindow(db, 'industry', t, true, true);
|
|
row = (await db.query(`SELECT * FROM "SendingMonitorSnapshot" WHERE "dimensionKey"=$1`, [key])).rows[0];
|
|
assert.equal(row.revision, 2);
|
|
assert.deepEqual(
|
|
row.metrics.metrics.map((m) => [m.success, m.mature, m.observing]),
|
|
[
|
|
[2, 4, 0],
|
|
[3, 4, 0],
|
|
[4, 4, 0],
|
|
],
|
|
);
|
|
checks++;
|
|
assert.equal(
|
|
(await db.query(`SELECT count(*)::int n FROM "SendingMonitorAlert" WHERE state='active'`)).rows[0].n,
|
|
1,
|
|
);
|
|
checks++;
|
|
await evaluateWindow(db, 'industry', t, true, true);
|
|
assert.equal(
|
|
(await db.query(`SELECT revision FROM "SendingMonitorSnapshot" WHERE "dimensionKey"=$1`, [key])).rows[0].revision,
|
|
2,
|
|
);
|
|
checks++;
|
|
await evaluateWindow(db, 'verification', t, true, true);
|
|
const business = (await db.query(`SELECT metrics,status FROM "SendingMonitorSnapshot" WHERE type='verification'`))
|
|
.rows[0];
|
|
assert.equal(business.metrics.total, 4);
|
|
assert.equal(business.status, 'unconfigured');
|
|
checks++;
|
|
|
|
// Removing enrollment in a later period cannot rewrite a historical window.
|
|
await db.query(`INSERT INTO "SendingMonitorTargetVersion" VALUES($1,2,false,$2,'qa')`, [
|
|
channel,
|
|
new Date(t.getTime() + 300000),
|
|
]);
|
|
await evaluateWindow(db, 'industry', t, true, true);
|
|
assert.equal(
|
|
(await db.query(`SELECT metrics FROM "SendingMonitorSnapshot" WHERE "dimensionKey"=$1`, [key])).rows[0].metrics
|
|
.total,
|
|
4,
|
|
);
|
|
checks++;
|
|
await db.query(`UPDATE "SmsSubmitRecord" SET "receiptRequested"=false WHERE id='qa-submit-0'`);
|
|
await projectMessages(db, ['qa-message-0']);
|
|
assert.equal(
|
|
(await db.query(`SELECT reason FROM "SendingMonitorFact" WHERE id='attempt:qa-submit-0'`)).rows[0].reason,
|
|
'receipt_not_requested',
|
|
);
|
|
checks++;
|
|
await db.query(`UPDATE "SmsSubmitRecord" SET "receiptRequested"=true WHERE id='qa-submit-0'`);
|
|
await db.query(`UPDATE "UpstreamReceiptInbox" SET "gatewayReceivedAt"=NULL WHERE id='qa-receipt-0'`);
|
|
await projectMessages(db, ['qa-message-0']);
|
|
assert.equal(
|
|
(await db.query(`SELECT reason FROM "SendingMonitorFact" WHERE id='attempt:qa-submit-0'`)).rows[0].reason,
|
|
'missing_receipt_time',
|
|
);
|
|
checks++;
|
|
const created = (await db.query(`SELECT min("createdAt") at FROM "ReportReadinessEvent"`)).rows[0].at;
|
|
assert(
|
|
Math.abs(created.getTime() - Date.now()) < 60000,
|
|
'Notification timestamps must remain UTC under Asia/Shanghai session',
|
|
);
|
|
checks++;
|
|
const drainage = (
|
|
await db.query(
|
|
`SELECT id,"signatureId","tenantId" FROM "SmsDrainageInfo" WHERE "signatureId" IS NOT NULL LIMIT 1`,
|
|
)
|
|
).rows[0];
|
|
if (drainage) {
|
|
const before = (
|
|
await db.query(`SELECT count(*)::int n FROM "ReportReadinessEvent" WHERE "reportType"='drainage'`)
|
|
).rows[0].n;
|
|
await db.query(
|
|
`INSERT INTO "ChannelSignatureReportTask" (id,"tenantId","signatureId","drainageItemId","reportType","channelId",carrier,"approvalScope",status,"updatedAt") VALUES($1,$2,$3,$4,'drainage',$5,NULL,'legacy_channel','approved',CURRENT_TIMESTAMP)`,
|
|
['qa-drainage', drainage.tenantId, drainage.signatureId, drainage.id, channel],
|
|
);
|
|
assert.equal(
|
|
(await db.query(`SELECT count(*)::int n FROM "ReportReadinessEvent" WHERE "reportType"='drainage'`)).rows[0].n,
|
|
before + 1,
|
|
);
|
|
checks++;
|
|
}
|
|
await db.query('ROLLBACK');
|
|
const leftover = (await db.query(`SELECT count(*)::int n FROM pg_namespace WHERE nspname=$1`, [schema])).rows[0].n;
|
|
assert.equal(leftover, 0);
|
|
checks++;
|
|
console.log(
|
|
JSON.stringify({
|
|
passed: checks,
|
|
database: 'real PostgreSQL',
|
|
isolation: 'transaction schema rolled back',
|
|
smsSent: 0,
|
|
}),
|
|
);
|
|
} catch (error) {
|
|
await db.query('ROLLBACK');
|
|
throw error;
|
|
} finally {
|
|
await db.end();
|
|
}
|
|
}
|
|
main().catch((error) => {
|
|
console.error(error.message);
|
|
process.exitCode = 1;
|
|
});
|