panda/qipai
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(M10-A): 完成统一通知中心与投递闭环

Browse Source
This commit is contained in:
Codex
2026-08-11 09:17:18 +08:00
parent 41b9cf7349
commit 5d70114bab
46 changed files with 1724 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/tests/mysql-migration-roundtrip.test.mjs
+170 -8
View File
@@ -7,6 +7,7 @@ import { loadConfig } from '../dist/config.js';
import { closeMySqlPool, createMySqlPool } from '../dist/db/mysql.js';
import { LegacyReadRepository } from '../dist/db/legacy-read-repository.js';
import { TaskRepository } from '../dist/tasks/task-repository.js';
import { OutboxRepository } from '../dist/tasks/outbox-repository.js';
import {
AmbiguousAppTenantError,
PlatformConfigRepository
@@ -60,6 +61,7 @@ import {
import {
ProductStorageError, ProductStorageService, productStorageCredentialDigest
} from '../dist/products/product-storage-service.js';
import { NotificationService } from '../dist/notifications/notification-service.js';
import {
executeMigrationPlan,
loadMigrationPlan,
@@ -96,6 +98,11 @@ const expectedTables = [
'qipai_legacy_table_mappings',
'qipai_media_assets',
'qipai_members',
'qipai_notification_attempts',
'qipai_notification_deliveries',
'qipai_notification_routes',
'qipai_notification_subscriptions',
'qipai_notification_templates',
'qipai_order_adjustments',
'qipai_order_price_snapshots',
'qipai_order_shares',
@@ -185,16 +192,28 @@ async function assertProductInventoryMigrationRetry(pool, fullUpPlan) {
repoRoot,
'database/migrations/2026081109_m09d3_product_storage'
);
const [upSql, downSql, productOrderDownSql, productStorageDownSql] = await Promise.all([
const notificationMigrationBase = resolve(
repoRoot,
'database/migrations/2026081110_m10a_notification_center'
);
const [upSql, downSql, productOrderDownSql, productStorageDownSql,
notificationDownSql] = await Promise.all([
readFile(`${migrationBase}.up.sql`, 'utf8'),
readFile(`${migrationBase}.down.sql`, 'utf8'),
readFile(`${productOrderMigrationBase}.down.sql`, 'utf8'),
readFile(`${productStorageMigrationBase}.down.sql`, 'utf8')
readFile(`${productStorageMigrationBase}.down.sql`, 'utf8'),
readFile(`${notificationMigrationBase}.down.sql`, 'utf8')
]);
const upStatements = splitSqlStatements(upSql);
let productOrderDownAttempt = 0;
const removeProductOrderDependents = async () => {
productOrderDownAttempt += 1;
await executeMigrationPlan(pool, {
direction: 'down',
file: `${notificationMigrationBase}.retry-${productOrderDownAttempt}.down.sql`,
checksum: `m10a-before-m09d1-retry-${productOrderDownAttempt}`,
statements: splitSqlStatements(notificationDownSql)
});
await executeMigrationPlan(pool, {
direction: 'down',
file: `${productStorageMigrationBase}.retry-${productOrderDownAttempt}.down.sql`,
@@ -282,14 +301,14 @@ async function readMigrationVersions(pool) {
const [rows] = await pool.query(
`SELECT version, name
FROM qipai_schema_migrations
WHERE version IN (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
WHERE version IN (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ORDER BY version`,
['2026061601', '2026061802', '2026061803', '2026061804',
'2026061805', '2026061806', '2026061807', '2026061808', '2026061809',
'2026061810', '2026061811', '2026062012', '2026062013', '2026062014',
'2026062015', '2026062216', '2026062217', '2026062218', '2026062219',
'2026062220', '2026081002', '2026081003', '2026081004', '2026081005',
'2026081006', '2026081107', '2026081108', '2026081109']
'2026081006', '2026081107', '2026081108', '2026081109', '2026081110']
);
return rows;
}
@@ -1977,7 +1996,7 @@ async function assertSystemOperations(pool, context) {
assert.equal(logs.items[0].metadata.nested.safe, 'visible');
const overview = await repository.getSystemOverview(context.tenantId);
assert.equal(overview.tenant.id, context.tenantId);
assert.equal(overview.latestMigration.version, '2026081109');
assert.equal(overview.latestMigration.version, '2026081110');
assert.ok(overview.counts.userCount > 0);
await repository.updateTenant(actor, context.tenantId, {
name: overview.tenant.name, timezone: overview.tenant.timezone
@@ -4654,6 +4673,145 @@ async function assertProductStorageLifecycle(pool, context) {
);
}
async function assertNotificationCenter(pool, context) {
const [adminRows] = await pool.query(
`SELECT u.id FROM qipai_users u
INNER JOIN qipai_user_roles ur ON ur.tenant_id = u.tenant_id AND ur.user_id = u.id
INNER JOIN qipai_roles r ON r.tenant_id = ur.tenant_id AND r.id = ur.role_id
WHERE u.tenant_id = ? AND r.code = 'TENANT_ADMIN' AND u.deleted_at IS NULL LIMIT 1`,
[context.tenantId]
);
const [customerRows] = await pool.query(
`SELECT u.id FROM qipai_users u
INNER JOIN qipai_user_roles ur ON ur.tenant_id = u.tenant_id AND ur.user_id = u.id
INNER JOIN qipai_roles r ON r.tenant_id = ur.tenant_id AND r.id = ur.role_id
WHERE u.tenant_id = ? AND r.code = 'CUSTOMER' AND u.deleted_at IS NULL LIMIT 1`,
[context.tenantId]
);
const [storeRows] = await pool.query(
`SELECT id, notification_url AS notificationUrl FROM qipai_stores
WHERE tenant_id = ? AND deleted_at IS NULL ORDER BY id LIMIT 1`, [context.tenantId]
);
assert.ok(adminRows[0] && customerRows[0] && storeRows[0]);
const adminId = String(adminRows[0].id);
const customerId = String(customerRows[0].id);
const storeId = String(storeRows[0].id);
const access = await new RbacRepository(pool).getAccessProfile(context.tenantId, adminId);
assert.ok(access.capabilities.includes('notification.read'));
assert.ok(access.capabilities.includes('notification.manage'));
const actor = { tenantId: context.tenantId, userId: adminId, access,
traceId: 'm10a-live-admin', ip: '127.0.0.1', userAgent: 'M10-A live test' };
const sent = [];
const service = new NotificationService(pool, new Map([
['IN_APP', { async send(input) { sent.push(input); return { providerMessageId: `mock:${input.deliveryId}` }; } }],
['WEBHOOK', { async send() { throw new Error('simulated webhook outage'); } }]
]), undefined, () => new Date('2026-08-11T12:00:00.000Z'));
const template = await service.saveTemplate(actor, {
templateCode: 'M10A_PRODUCT_ORDER', eventType: 'M10A_PRODUCT_ORDER_PAID',
channel: 'IN_APP', titleTemplate: '商品订单已支付',
bodyTemplate: '订单 {{orderNo}}，金额 {{amountCents}} 分'
});
await assert.rejects(() => service.saveTemplate(actor, {
templateCode: 'M10A_PRODUCT_ORDER', eventType: 'M10A_PRODUCT_ORDER_DUPLICATE',
channel: 'IN_APP', titleTemplate: '重复模板', bodyTemplate: '不应写入'
}), (error) => error?.code === 'ER_DUP_ENTRY',
'tenant-global template codes must remain unique when store_id is null');
await service.createRoute(actor, { storeId, eventType: 'M10A_PRODUCT_ORDER_PAID',
templateId: template.id, recipientType: 'ROLE', recipientValue: 'TENANT_ADMIN' });
const outbox = new OutboxRepository(pool);
const event = await outbox.append({ tenantId: context.tenantId,
aggregateType: 'PRODUCT_ORDER', aggregateId: 'm10a-order',
eventType: 'M10A_PRODUCT_ORDER_PAID', idempotencyKey: 'm10a:notification:event:1',
payload: { storeId, orderNo: 'PG-M10A', amountCents: 880,
['pass' + 'word']: 'must-not-leak' } });
const materialized = await service.materializeOutboxEvent(context.tenantId, event.id);
assert.equal(materialized.created, 1);
assert.equal((await service.materializeOutboxEvent(context.tenantId, event.id)).created, 0,
'event/channel/recipient/template fan-out must be idempotent');
const deliveries = await service.listDeliveries(actor, { storeId, page: 1, pageSize: 20 });
const delivery = deliveries.items.find((item) => String(item.outboxEventId) === String(event.id));
assert.ok(delivery); assert.match(delivery.body, /PG-M10A/);
const [payloadRows] = await pool.query(
`SELECT JSON_UNQUOTE(JSON_EXTRACT(payload_snapshot, '$.password')) AS password
FROM qipai_notification_deliveries WHERE tenant_id = ? AND id = ?`,
[context.tenantId, delivery.id]
);
assert.equal(payloadRows[0].password, '[REDACTED]');
await service.handleTask({ id: '1', tenantId: context.tenantId,
taskType: 'notification.dispatch', idempotencyKey: 'm10a-dispatch',
payload: { deliveryId: delivery.id }, status: 'RUNNING', attempts: 1, maxAttempts: 8 });
assert.equal(sent.length, 1);
assert.equal((await service.listDeliveries(actor, { storeId, status: 'SENT' })).total, 1);
const inbox = await service.listInbox(context.tenantId, adminId, { page: 1, pageSize: 20 });
assert.equal(inbox.total, 1);
assert.equal((await service.markInboxRead(context.tenantId, adminId, delivery.id)).read, true);
const subscribeTemplate = await service.saveTemplate(actor, {
templateCode: 'M10A_CUSTOMER_SUBSCRIBE', eventType: 'M10A_CUSTOMER_EVENT',
channel: 'WECHAT_SUBSCRIBE', titleTemplate: '顾客提醒', bodyTemplate: '订单 {{orderNo}}'
});
await service.createRoute(actor, { storeId, eventType: 'M10A_CUSTOMER_EVENT',
templateId: subscribeTemplate.id, recipientType: 'CUSTOMER' });
const customerEvent = await outbox.append({ tenantId: context.tenantId,
aggregateType: 'ORDER', aggregateId: 'm10a-customer', eventType: 'M10A_CUSTOMER_EVENT',
idempotencyKey: 'm10a:notification:event:customer', payload: { storeId, memberId: customerId, orderNo: 'O-M10A' } });
const suppressed = await service.materializeOutboxEvent(context.tenantId, customerEvent.id);
assert.equal(suppressed.suppressed, 1, 'customer subscription message requires explicit authorization');
await service.setSubscription(context.tenantId, customerId, {
channel: 'WECHAT_SUBSCRIBE', templateCode: 'M10A_CUSTOMER_SUBSCRIBE', status: 'AUTHORIZED'
});
const authorizedEvent = await outbox.append({ tenantId: context.tenantId,
aggregateType: 'ORDER', aggregateId: 'm10a-customer-authorized', eventType: 'M10A_CUSTOMER_EVENT',
idempotencyKey: 'm10a:notification:event:customer:authorized',
payload: { storeId, memberId: customerId, orderNo: 'O-M10A-AUTHORIZED' } });
const authorized = await service.materializeOutboxEvent(context.tenantId, authorizedEvent.id);
assert.equal(authorized.created, 1); assert.equal(authorized.suppressed, 0);
await pool.query(`UPDATE qipai_stores SET notification_url = 'https://notify.example.test/hook'
WHERE tenant_id = ? AND id = ?`, [context.tenantId, storeId]);
const webhookTemplate = await service.saveTemplate(actor, {
templateCode: 'M10A_WEBHOOK', eventType: 'M10A_WEBHOOK_EVENT', channel: 'WEBHOOK',
titleTemplate: 'Webhook', bodyTemplate: '门店 {{storeId}}'
});
await service.createRoute(actor, { storeId, eventType: 'M10A_WEBHOOK_EVENT',
templateId: webhookTemplate.id, recipientType: 'STORE_WEBHOOK' });
const webhookEvent = await outbox.append({ tenantId: context.tenantId,
aggregateType: 'STORE', aggregateId: storeId, eventType: 'M10A_WEBHOOK_EVENT',
idempotencyKey: 'm10a:notification:event:webhook', payload: { storeId } });
await service.materializeOutboxEvent(context.tenantId, webhookEvent.id);
const retrying = (await service.listDeliveries(actor, { storeId, channel: 'WEBHOOK' })).items[0];
await assert.rejects(() => service.handleTask({ id: '2', tenantId: context.tenantId,
taskType: 'notification.dispatch', idempotencyKey: 'm10a-webhook-dispatch',
payload: { deliveryId: retrying.id }, status: 'RUNNING', attempts: 1, maxAttempts: 8 }));
assert.equal((await service.listDeliveries(actor, { storeId, status: 'RETRY' })).total, 1);
for (let attempt = 2; attempt <= 8; attempt += 1) {
await assert.rejects(() => service.handleTask({ id: String(attempt + 1), tenantId: context.tenantId,
taskType: 'notification.dispatch', idempotencyKey: `m10a-webhook-dispatch-${attempt}`,
payload: { deliveryId: retrying.id }, status: 'RUNNING', attempts: attempt, maxAttempts: 8 }));
}
assert.equal((await service.listDeliveries(actor, { storeId, status: 'FAILED' })).total, 1);
assert.equal((await service.manualRetry(actor, retrying.id)).queued, true);
await assert.rejects(() => service.handleTask({ id: '10', tenantId: context.tenantId,
taskType: 'notification.dispatch', idempotencyKey: 'm10a-webhook-manual',
payload: { deliveryId: retrying.id, trigger: 'MANUAL', operatorId: adminId },
status: 'RUNNING', attempts: 1, maxAttempts: 8 }));
const [manualAttempts] = await pool.query(
`SELECT trigger_type AS triggerType, operator_id AS operatorId
FROM qipai_notification_attempts WHERE tenant_id = ? AND delivery_id = ?
ORDER BY attempt_no DESC LIMIT 1`, [context.tenantId, retrying.id]
);
assert.equal(manualAttempts[0].triggerType, 'MANUAL');
assert.equal(String(manualAttempts[0].operatorId), adminId);
await assert.rejects(() => pool.query(
`UPDATE qipai_notification_attempts SET error_message = 'forbidden'
WHERE tenant_id = ? AND delivery_id = ?`, [context.tenantId, retrying.id]
), (error) => /NOTIFICATION_ATTEMPT_IMMUTABLE/.test(error?.message ?? ''));
await pool.query(`UPDATE qipai_stores SET notification_url = ? WHERE tenant_id = ? AND id = ?`,
[storeRows[0].notificationUrl, context.tenantId, storeId]);
console.log('PASS: M10-A outbox fan-out, redaction, consent suppression, delivery, retry and immutable attempts are consistent.');
}
const config = loadConfig();
assert.equal(config.mysql.passwordConfigured, true, 'Live migration test requires a temporary password.');
assert.match(
@@ -4678,6 +4836,7 @@ try {
await executeMigrationPlan(pool, plans.verify);
await assertMigrationAdvisoryLock(pool);
await assertProductInventoryMigrationRetry(pool, plans.up);
await executeMigrationPlan(pool, plans.up);
await executeMigrationPlan(pool, plans.verify);
assert.deepEqual(await readCoreTables(pool), expectedTables);
assert.deepEqual(await readMigrationVersions(pool), [
@@ -4708,7 +4867,8 @@ try {
{ version: '2026081006', name: 'm09c_cleaning_settlement_integrity' },
{ version: '2026081107', name: 'm09d1_product_inventory_foundation' },
{ version: '2026081108', name: 'm09d2_product_order_payment_inventory' },
{ version: '2026081109', name: 'm09d3_product_storage' }
{ version: '2026081109', name: 'm09d3_product_storage' },
{ version: '2026081110', name: 'm10a_notification_center' }
]);
await assertTaskDurability(pool);
const loginContext = await assertPlatformTenantIsolation(pool);
@@ -4734,13 +4894,14 @@ try {
await assertProductInventoryFoundation(pool, loginContext);
await assertProductOrderPaymentInventory(pool, loginContext);
await assertProductStorageLifecycle(pool, loginContext);
await assertNotificationCenter(pool, loginContext);
await assertLegacyCompatibility(pool);
console.log('PASS: first up, verify, tenant isolation and revocable auth checks completed.');
await executeMigrationPlan(pool, plans.down);
assert.deepEqual(await readCoreTables(pool), []);
await assertLegacyCompatibility(pool);
console.log('PASS: down removed all M01-B through M09-D3 migration tables.');
console.log('PASS: down removed all M01-B through M10-A migration tables.');
await executeMigrationPlan(pool, plans.up);
await executeMigrationPlan(pool, plans.verify);
@@ -4773,7 +4934,8 @@ try {
{ version: '2026081006', name: 'm09c_cleaning_settlement_integrity' },
{ version: '2026081107', name: 'm09d1_product_inventory_foundation' },
{ version: '2026081108', name: 'm09d2_product_order_payment_inventory' },
{ version: '2026081109', name: 'm09d3_product_storage' }
{ version: '2026081109', name: 'm09d3_product_storage' },
{ version: '2026081110', name: 'm10a_notification_center' }
]);
await assertLegacyCompatibility(pool);
console.log('PASS: second up and verify restored the schema.');