feat(pricing): 开发定价方案

apps/worker/src/exchange-rate-sync.ts

@@ -0,0 +1,68 @@
+import type { Database } from '@atlas/database';
+
+/** CBR 的 Value 按 Nominal 报价；用整数除法换算，避免浮点误差与非预期 XML 内容。 */
+export function parseCbrCny(xml: string) {
+    const date = xml.match(/<ValCurs\b[^>]*Date="(\d{2})\.(\d{2})\.(\d{4})"/);
+    const entry = [...xml.matchAll(/<Valute\b[^>]*>([\s\S]*?)<\/Valute>/g)]
+        .map((m) => m[1])
+        .find((v) => /<CharCode>\s*CNY\s*<\/CharCode>/.test(v));
+    const nominal = entry?.match(/<Nominal>\s*(\d{1,6})\s*<\/Nominal>/)?.[1],
+        value = entry?.match(/<Value>\s*(\d{1,9}),?(\d{0,10})\s*<\/Value>/);
+    if (!date || !nominal || !value || BigInt(nominal) === 0n)
+        throw new Error('CBR 响应缺少有效人民币报价');
+    const effectiveOn = `${date[3]}-${date[2]}-${date[1]}`;
+    const parsedDate = new Date(`${effectiveOn}T00:00:00Z`);
+    if (
+        !Number.isFinite(parsedDate.getTime()) ||
+        parsedDate.toISOString().slice(0, 10) !== effectiveOn ||
+        parsedDate.getTime() > Date.now() + 2 * 86400000
+    )
+        throw new Error('CBR 报价日期无效');
+    const scaled =
+        BigInt(value[1]) * 10000000000n +
+        BigInt((value[2] ?? '').padEnd(10, '0'));
+    const rate = (scaled + BigInt(nominal) / 2n) / BigInt(nominal);
+    if (rate <= 0n) throw new Error('CBR 报价必须大于零');
+    const digits = rate.toString().padStart(11, '0');
+    return {
+        effectiveOn,
+        rate: `${digits.slice(0, -10)}.${digits.slice(-10)}`,
+    };
+}
+export async function syncExchangeRate(database: Database) {
+    await database.transaction(async (tx) => {
+        const lock = await tx.query<{ locked: boolean }>(
+            `SELECT pg_try_advisory_xact_lock(74103921) AS locked`,
+        );
+        if (!lock.rows[0].locked) return;
+        const recent = await tx.query(
+            `SELECT id FROM atlas.exchange_rates WHERE source='cbr' AND kind='official_reference' AND base_currency_code='CNY' AND quote_currency_code='RUB' AND fetched_at>now()-interval '2 hours' LIMIT 1`,
+        );
+        if (recent.rowCount) return;
+        const response = await fetch(
+            'https://www.cbr.ru/scripts/XML_daily.asp',
+            { signal: AbortSignal.timeout(15000) },
+        );
+        if (!response.ok)
+            throw new Error(`CBR 请求失败：HTTP ${response.status}`);
+        const bytes = await response.arrayBuffer();
+        if (bytes.byteLength > 1000000) throw new Error('CBR 响应大小异常');
+        const xml = new TextDecoder('windows-1251').decode(bytes),
+            rate = parseCbrCny(xml);
+        await tx.query(
+            `INSERT INTO atlas.exchange_rates(base_currency_code,quote_currency_code,rate,kind,source,source_reference,effective_on,observed_at) VALUES('CNY','RUB',$1,'official_reference','cbr','https://www.cbr.ru/scripts/XML_daily.asp',$2,now()) ON CONFLICT(base_currency_code,quote_currency_code,kind,source,effective_on) DO UPDATE SET fetched_at=now()`,
+            [rate.rate, rate.effectiveOn],
+        );
+    });
+}
+export async function cleanupPricingPreviews(database: Database) {
+    const workspaces = await database.query<{ id: string }>(
+        'SELECT id FROM atlas.workspaces',
+    );
+    for (const w of workspaces.rows)
+        await database.workspaceTransaction(w.id, async (tx) => {
+            await tx.query(
+                `DELETE FROM atlas.pricing_previews WHERE expires_at<now()-interval '2 days'`,
+            );
+        });
+}

apps/worker/src/main.ts

@@ -1,3 +1,4 @@
+import { syncExchangeRate, cleanupPricingPreviews } from './exchange-rate-sync.js';
 import { randomUUID } from 'node:crypto';
 import { S3Client } from '@aws-sdk/client-s3';
 import { Queue, Worker } from 'bullmq';
@@ -460,6 +461,15 @@ async function bootstrap() {
         'Atlas Worker is ready; consuming transactional-outbox events.',
     );
 }
+// 汇率与预览清理可重复执行，失败保留上次汇率，不影响其他业务队列。
+const pricingMaintenance=async()=>{
+    await Promise.allSettled([
+        syncExchangeRate(database).catch(()=>console.error('pricing_exchange_rate_sync_failed')),
+        cleanupPricingPreviews(database).catch(()=>console.error('pricing_preview_cleanup_failed')),
+    ]);
+};
+const pricingTimer=setInterval(()=>void pricingMaintenance(),2*60*60_000);
+void pricingMaintenance();
 void bootstrap().catch((error: unknown) => {
     console.error(
         'worker_startup_failed',
@@ -468,6 +478,7 @@ void bootstrap().catch((error: unknown) => {
     process.exitCode = 1;
 });
 async function shutdown() {
+    clearInterval(pricingTimer);
     if (dispatcherTimer) clearInterval(dispatcherTimer);
     if (mediaCleanupTimer) clearInterval(mediaCleanupTimer);
     if (platformCatalogTimer) clearInterval(platformCatalogTimer);