done-rate 统计优化

oms/OPTIMIZATION_TRACKER.md

@@ -31,6 +31,7 @@
 - 处理:
   - 已改为 1 条聚合查询，使用 `uniqIf` 和 `countIf` 一次返回三类统计结果。
   - 时间边界已修正为 `[day_start, next_day_start)`，避免漏掉当天末尾数据。
+  - `DoneRate` 每日落库阶段已从逐条 `createOrUpdate` 改为分批 `bulkWrite upsert`（按 `date + res`），减少 Mongo 写入往返。
 
 ### 3. `done-rate` 增加 ClickHouse 查询耗时与结果规模日志
 
@@ -55,17 +56,20 @@
 
 ## 待修正
 
-### 1. `done-rate` 后半段 Mongo 聚合更新可能仍然偏慢
+### 1. `done-rate` 后半段 Mongo 聚合更新
 
-- 状态: `待修正`
+- 状态: `已修正`
 - 文件: `services/cron-jobs/done-rate.ts`
-- 现状:
-  - ClickHouse 查询已经优化，但后半段仍然会聚合 `DoneRateModel` 并逐条更新 `TotalDoneRate`。
-  - 当作品数量继续增长时，这一段可能成为新的耗时热点。
-- 可选优化方向:
-  - 改为 `bulkWrite` 批量更新 `TotalDoneRate`。
-  - 评估是否可以用增量方式替代全量聚合。
-  - 增加阶段性耗时日志，拆分 ClickHouse 时间和 Mongo 更新时间。
+- 原问题:
+  - ClickHouse 查询优化后，后半段仍逐条 `findByIdAndUpdate` 更新 `TotalDoneRate`，网络往返与写入放大明显。
+- 处理:
+  - 已改为分批 `bulkWrite`（`TOTAL_DONE_RATE_BULK_SIZE=1000`）覆盖更新 + `upsert`。
+  - 增加阶段性日志，拆分为：
+    - `DoneRateModel.aggregate` 聚合耗时
+    - `TotalDoneRate.bulkWrite` 写入耗时与批次进度
+- 结果:
+  - 显著减少逐条更新带来的 Mongo 往返开销。
+  - 后续可更准确判断瓶颈位于聚合阶段还是写入阶段。
 
 ### 2. `ingestor-service` 先 `ack` 后 flush，存在小窗口数据丢失风险
 

oms/dist/services/cron-jobs/done-rate.js

@@ -3,13 +3,14 @@ var __importDefault = (this && this.__importDefault) || function (mod) {
     return (mod && mod.__esModule) ? mod : { "default": mod };
 };
 const dayjs_1 = __importDefault(require("dayjs"));
-const doneRateService_1 = __importDefault(require("../../src/services/doneRateService")); // 导入 DoneRateService
 const clients_1 = require("../../src/services/clients");
-const mongoose_1 = __importDefault(require("mongoose")); // 导入 mongoose 和 Connection 用于处理远程连接
+const mongoose_1 = __importDefault(require("mongoose"));
 const totalDoneRateModel_1 = __importDefault(require("../../src/models/totalDoneRateModel")); // 导入 TotalDoneRate 模型 (已包含 totalTipCount)
 const doneRateModel_1 = __importDefault(require("../../src/models/doneRateModel")); // 导入 DoneRateModel 用于聚合历史数据
 // ClickHouse 表名
 const CLICKHOUSE_EVENTS_TABLE = "events"; // 确保与 ClickHouseService 中的表名一致
+const TOTAL_DONE_RATE_BULK_SIZE = 1000;
+const DAILY_DONE_RATE_BULK_SIZE = 1000;
 /**
  * 每日统计作品完成率。
  * 如果提供了 dateStr (YYYY-MM-DD 或 YYYYMMDD)，则统计该日的数据；否则，默认统计昨天的数据。
@@ -85,27 +86,49 @@ async function run(dateStr) {
             totalTipEvents += Number(row.tip_count) || 0;
         });
         console.log(`[DoneRate Cron] ClickHouse aggregate query completed in ${clickhouseQueryElapsedMs}ms. Rows: ${aggregateResults.length}, valid artworks: ${artworkStartCounts.size}, invalid rows: ${invalidArtworkRowCount}, unique starts: ${totalStartUsers}, unique dones: ${totalDoneUsers}, tip events: ${totalTipEvents}.`);
-        // --- 2. 合并数据并更新 DoneRate 模型 (每日记录) ---
+        // --- 2. 合并数据并批量更新 DoneRate 模型 (每日记录) ---
         let updatedRecordsCount = 0;
         let createdRecordsCount = 0;
         // 获取所有需要处理的作品 ID 集合 (Start + Done + Tip)
-        const allResIds = new Set([...artworkStartCounts.keys(), ...artworkDoneCounts.keys(), ...artworkTipCounts.keys()]);
-        for (const resIdStr of allResIds.values()) {
-            const startCount = artworkStartCounts.get(resIdStr) || 0;
-            const doneCount = artworkDoneCounts.get(resIdStr) || 0;
-            const tipCount = artworkTipCounts.get(resIdStr) || 0;
-            const resObjectId = new mongoose_1.default.Types.ObjectId(resIdStr);
-            // 使用 DoneRateService 来创建或更新记录，并传入 tipCount
-            const doneRateDoc = await doneRateService_1.default.createOrUpdateDoneRate(yesterdayYYYYMMDD, resObjectId, startCount, doneCount, tipCount);
-            if (doneRateDoc.createdAt.getTime() === doneRateDoc.updatedAt.getTime()) {
-                createdRecordsCount++;
-            }
-            else {
-                updatedRecordsCount++;
-            }
+        const allResIds = [...new Set([...artworkStartCounts.keys(), ...artworkDoneCounts.keys(), ...artworkTipCounts.keys()])];
+        const doneRateBulkStartedAt = Date.now();
+        console.log(`[DoneRate Cron] 开始通过 bulkWrite 更新每日 DoneRate。总计 ${allResIds.length} 个作品，batchSize=${DAILY_DONE_RATE_BULK_SIZE}。`);
+        for (let i = 0; i < allResIds.length; i += DAILY_DONE_RATE_BULK_SIZE) {
+            const chunk = allResIds.slice(i, i + DAILY_DONE_RATE_BULK_SIZE);
+            const ops = chunk.map((resIdStr) => {
+                const startCount = artworkStartCounts.get(resIdStr) || 0;
+                const doneCount = artworkDoneCounts.get(resIdStr) || 0;
+                const tipCount = artworkTipCounts.get(resIdStr) || 0;
+                const completionRate = startCount > 0 ? (doneCount / startCount) * 100 : 0;
+                const resObjectId = new mongoose_1.default.Types.ObjectId(resIdStr);
+                return {
+                    updateOne: {
+                        filter: { date: yesterdayYYYYMMDD, res: resObjectId },
+                        update: {
+                            $set: {
+                                startCount,
+                                doneCount,
+                                tipCount,
+                                completionRate,
+                            },
+                            $setOnInsert: {
+                                date: yesterdayYYYYMMDD,
+                                res: resObjectId,
+                            },
+                        },
+                        upsert: true,
+                    },
+                };
+            });
+            const result = await doneRateModel_1.default.bulkWrite(ops, { ordered: false });
+            createdRecordsCount += result.upsertedCount || 0;
+            updatedRecordsCount += chunk.length - (result.upsertedCount || 0);
+            const processed = Math.min(i + DAILY_DONE_RATE_BULK_SIZE, allResIds.length);
+            const elapsed = Date.now() - doneRateBulkStartedAt;
+            console.log(`[DoneRate Cron] 每日DoneRate进度: 已处理 ${processed}/${allResIds.length} 条，耗时 ${elapsed}ms`);
         }
         const totalProcessedArtworks = createdRecordsCount + updatedRecordsCount;
-        console.log(`[DoneRate Cron] DoneRate model (Daily) update completed. Total artworks processed: ${totalProcessedArtworks}. Created: ${createdRecordsCount}, Updated: ${updatedRecordsCount}.`);
+        console.log(`[DoneRate Cron] DoneRate model (Daily) update completed. Total artworks processed: ${totalProcessedArtworks}. Created: ${createdRecordsCount}, Updated: ${updatedRecordsCount}. Elapsed: ${Date.now() - doneRateBulkStartedAt}ms.`);
         // --- 3. 重新聚合并更新本地的 TotalDoneRate 表 (累计记录) ---
         // 通过聚合 DoneRate 日记录来计算最新的累计总数，确保幂等性。
         // 3.1 聚合所有历史 DoneRate 记录直到昨天 (即 targetDay)
@@ -121,43 +144,44 @@ async function run(dateStr) {
                 },
             },
         ];
+        const totalAggregateStartedAt = Date.now();
         const aggregatedTotals = await doneRateModel_1.default.aggregate(aggregationPipeline);
+        const totalAggregateElapsedMs = Date.now() - totalAggregateStartedAt;
+        console.log(`[DoneRate Cron] TotalDoneRate 聚合完成。作品数 ${aggregatedTotals.length}，聚合耗时 ${totalAggregateElapsedMs}ms。`);
+        const totalUpdateStartedAt = Date.now();
         let updatedTotalDoneRateCount = 0;
-        let updateStartTime = new Date().getTime();
-        console.log(`[DoneRate Cron] 开始通过聚合更新本地 TotalDoneRate 表。总计 ${aggregatedTotals.length} 个作品的累计数据。`);
-        // 3.2 遍历聚合结果，覆盖式更新 TotalDoneRate
-        for (let i = 0; i < aggregatedTotals.length; i++) {
-            const totalData = aggregatedTotals[i];
-            const artworkId = totalData._id;
-            const { totalStartCount, totalDoneCount, totalTipCount } = totalData;
-            // 1. 计算总完成率
-            const newCompletionRate = totalStartCount > 0 ? (totalDoneCount / totalStartCount) * 100 : 0;
-            try {
-                // 2. 使用 findByIdAndUpdate 进行原子操作 (覆盖式更新)
-                const updatedDoc = await totalDoneRateModel_1.default.findByIdAndUpdate(artworkId, {
-                    $set: {
-                        totalStartCount: totalStartCount,
-                        totalDoneCount: totalDoneCount,
-                        totalTipCount: totalTipCount, // 写入新的累计道具使用次数
-                        completionRate: newCompletionRate,
+        console.log(`[DoneRate Cron] 开始通过 bulkWrite 更新本地 TotalDoneRate 表。总计 ${aggregatedTotals.length} 个作品，batchSize=${TOTAL_DONE_RATE_BULK_SIZE}。`);
+        for (let i = 0; i < aggregatedTotals.length; i += TOTAL_DONE_RATE_BULK_SIZE) {
+            const chunk = aggregatedTotals.slice(i, i + TOTAL_DONE_RATE_BULK_SIZE);
+            const ops = chunk.map((totalData) => {
+                const artworkId = totalData._id;
+                const totalStartCount = Number(totalData.totalStartCount) || 0;
+                const totalDoneCount = Number(totalData.totalDoneCount) || 0;
+                const totalTipCount = Number(totalData.totalTipCount) || 0;
+                const completionRate = totalStartCount > 0 ? (totalDoneCount / totalStartCount) * 100 : 0;
+                return {
+                    updateOne: {
+                        filter: { _id: artworkId },
+                        update: {
+                            $set: {
+                                totalStartCount,
+                                totalDoneCount,
+                                totalTipCount,
+                                completionRate,
+                            },
+                        },
+                        upsert: true,
                     },
-                }, { new: true, upsert: true } // upsert: true 表示如果文档不存在则创建
-                );
-                if (updatedDoc) {
-                    updatedTotalDoneRateCount++;
-                }
-                // 进度日志
-                if ((i + 1) % 50 === 0 || i === aggregatedTotals.length - 1) {
-                    const elapsed = new Date().getTime() - updateStartTime;
-                    console.log(`[DoneRate Cron] 进度: 已更新 ${i + 1}/${aggregatedTotals.length} 条本地 TotalDoneRate 记录, 当前耗时: ${elapsed}ms`);
-                }
-            }
-            catch (totalUpdateError) {
-                console.error(`[DoneRate Cron] Error updating TotalDoneRate document for artwork ID ${artworkId}:`, totalUpdateError);
-            }
+                };
+            });
+            await totalDoneRateModel_1.default.bulkWrite(ops, { ordered: false });
+            updatedTotalDoneRateCount += chunk.length;
+            const processed = Math.min(i + TOTAL_DONE_RATE_BULK_SIZE, aggregatedTotals.length);
+            const elapsed = Date.now() - totalUpdateStartedAt;
+            console.log(`[DoneRate Cron] 进度: 已处理 ${processed}/${aggregatedTotals.length} 条 TotalDoneRate，累计耗时 ${elapsed}ms`);
         }
-        const updateTimeTaken = new Date().getTime() - updateStartTime;
-        console.log(`[DoneRate Cron] 本地 TotalDoneRate 表更新完成。总计更新 ${updatedTotalDoneRateCount} 条记录，总耗时 ${updateTimeTaken}ms。`);
+        const totalUpdateElapsedMs = Date.now() - totalUpdateStartedAt;
+        console.log(`[DoneRate Cron] 本地 TotalDoneRate 表更新完成。处理记录数 ${aggregatedTotals.length}，写入阶段耗时 ${totalUpdateElapsedMs}ms。`);
         const summary = `[DoneRate Cron] Daily done-rate calculation for ${yesterdayYYYYMMDD} completed. Total DoneRate processed: ${totalProcessedArtworks}. Created DoneRate: ${createdRecordsCount}, Updated DoneRate: ${updatedRecordsCount}. Updated TotalDoneRate records: ${updatedTotalDoneRateCount}.`;
         console.log(summary);
         return summary;

oms/services/cron-jobs/done-rate.ts

@@ -1,7 +1,6 @@
 import dayjs from "dayjs";
-import doneRateService from "../../src/services/doneRateService"; // 导入 DoneRateService
 import { clickhouseService } from "../../src/services/clients";
-import mongoose, { Connection } from "mongoose"; // 导入 mongoose 和 Connection 用于处理远程连接
+import mongoose from "mongoose";
 import TotalDoneRate from "../../src/models/totalDoneRateModel"; // 导入 TotalDoneRate 模型 (已包含 totalTipCount)
 import DoneRateModel from "../../src/models/doneRateModel"; // 导入 DoneRateModel 用于聚合历史数据
 
@@ -18,6 +17,9 @@ interface ClickHouseDoneRateAggregateResult {
   tip_count: number; // 道具使用次数
 }
 
+const TOTAL_DONE_RATE_BULK_SIZE = 1000;
+const DAILY_DONE_RATE_BULK_SIZE = 1000;
+
 /**
  * 每日统计作品完成率。
  * 如果提供了 dateStr (YYYY-MM-DD 或 YYYYMMDD)，则统计该日的数据；否则，默认统计昨天的数据。
@@ -104,31 +106,61 @@ async function run(dateStr?: string | dayjs.Dayjs): Promise<string> {
       `[DoneRate Cron] ClickHouse aggregate query completed in ${clickhouseQueryElapsedMs}ms. Rows: ${aggregateResults.length}, valid artworks: ${artworkStartCounts.size}, invalid rows: ${invalidArtworkRowCount}, unique starts: ${totalStartUsers}, unique dones: ${totalDoneUsers}, tip events: ${totalTipEvents}.`
     );
 
-    // --- 2. 合并数据并更新 DoneRate 模型 (每日记录) ---
+    // --- 2. 合并数据并批量更新 DoneRate 模型 (每日记录) ---
     let updatedRecordsCount = 0;
     let createdRecordsCount = 0;
 
     // 获取所有需要处理的作品 ID 集合 (Start + Done + Tip)
-    const allResIds = new Set([...artworkStartCounts.keys(), ...artworkDoneCounts.keys(), ...artworkTipCounts.keys()]);
-
-    for (const resIdStr of allResIds.values()) {
-      const startCount = artworkStartCounts.get(resIdStr) || 0;
-      const doneCount = artworkDoneCounts.get(resIdStr) || 0;
-      const tipCount = artworkTipCounts.get(resIdStr) || 0;
-      const resObjectId = new mongoose.Types.ObjectId(resIdStr);
-
-      // 使用 DoneRateService 来创建或更新记录，并传入 tipCount
-      const doneRateDoc = await doneRateService.createOrUpdateDoneRate(yesterdayYYYYMMDD, resObjectId, startCount, doneCount, tipCount);
-      if (doneRateDoc.createdAt.getTime() === doneRateDoc.updatedAt.getTime()) {
-        createdRecordsCount++;
-      } else {
-        updatedRecordsCount++;
-      }
+    const allResIds = [...new Set([...artworkStartCounts.keys(), ...artworkDoneCounts.keys(), ...artworkTipCounts.keys()])];
+
+    const doneRateBulkStartedAt = Date.now();
+    console.log(
+      `[DoneRate Cron] 开始通过 bulkWrite 更新每日 DoneRate。总计 ${allResIds.length} 个作品，batchSize=${DAILY_DONE_RATE_BULK_SIZE}。`
+    );
+
+    for (let i = 0; i < allResIds.length; i += DAILY_DONE_RATE_BULK_SIZE) {
+      const chunk = allResIds.slice(i, i + DAILY_DONE_RATE_BULK_SIZE);
+      const ops = chunk.map((resIdStr) => {
+        const startCount = artworkStartCounts.get(resIdStr) || 0;
+        const doneCount = artworkDoneCounts.get(resIdStr) || 0;
+        const tipCount = artworkTipCounts.get(resIdStr) || 0;
+        const completionRate = startCount > 0 ? (doneCount / startCount) * 100 : 0;
+        const resObjectId = new mongoose.Types.ObjectId(resIdStr);
+
+        return {
+          updateOne: {
+            filter: { date: yesterdayYYYYMMDD, res: resObjectId },
+            update: {
+              $set: {
+                startCount,
+                doneCount,
+                tipCount,
+                completionRate,
+              },
+              $setOnInsert: {
+                date: yesterdayYYYYMMDD,
+                res: resObjectId,
+              },
+            },
+            upsert: true,
+          },
+        };
+      });
+
+      const result = await DoneRateModel.bulkWrite(ops, { ordered: false });
+      createdRecordsCount += result.upsertedCount || 0;
+      updatedRecordsCount += chunk.length - (result.upsertedCount || 0);
+
+      const processed = Math.min(i + DAILY_DONE_RATE_BULK_SIZE, allResIds.length);
+      const elapsed = Date.now() - doneRateBulkStartedAt;
+      console.log(`[DoneRate Cron] 每日DoneRate进度: 已处理 ${processed}/${allResIds.length} 条，耗时 ${elapsed}ms`);
     }
 
     const totalProcessedArtworks = createdRecordsCount + updatedRecordsCount;
 
-    console.log(`[DoneRate Cron] DoneRate model (Daily) update completed. Total artworks processed: ${totalProcessedArtworks}. Created: ${createdRecordsCount}, Updated: ${updatedRecordsCount}.`);
+    console.log(
+      `[DoneRate Cron] DoneRate model (Daily) update completed. Total artworks processed: ${totalProcessedArtworks}. Created: ${createdRecordsCount}, Updated: ${updatedRecordsCount}. Elapsed: ${Date.now() - doneRateBulkStartedAt}ms.`
+    );
 
     // --- 3. 重新聚合并更新本地的 TotalDoneRate 表 (累计记录) ---
     // 通过聚合 DoneRate 日记录来计算最新的累计总数，确保幂等性。
@@ -147,53 +179,58 @@ async function run(dateStr?: string | dayjs.Dayjs): Promise<string> {
       },
     ];
 
+    const totalAggregateStartedAt = Date.now();
     const aggregatedTotals = await DoneRateModel.aggregate(aggregationPipeline);
+    const totalAggregateElapsedMs = Date.now() - totalAggregateStartedAt;
+
+    console.log(
+      `[DoneRate Cron] TotalDoneRate 聚合完成。作品数 ${aggregatedTotals.length}，聚合耗时 ${totalAggregateElapsedMs}ms。`
+    );
+
+    const totalUpdateStartedAt = Date.now();
     let updatedTotalDoneRateCount = 0;
-    let updateStartTime = new Date().getTime();
-
-    console.log(`[DoneRate Cron] 开始通过聚合更新本地 TotalDoneRate 表。总计 ${aggregatedTotals.length} 个作品的累计数据。`);
-
-    // 3.2 遍历聚合结果，覆盖式更新 TotalDoneRate
-    for (let i = 0; i < aggregatedTotals.length; i++) {
-      const totalData = aggregatedTotals[i];
-      const artworkId = totalData._id;
-
-      const { totalStartCount, totalDoneCount, totalTipCount } = totalData;
-
-      // 1. 计算总完成率
-      const newCompletionRate = totalStartCount > 0 ? (totalDoneCount / totalStartCount) * 100 : 0;
-
-      try {
-        // 2. 使用 findByIdAndUpdate 进行原子操作 (覆盖式更新)
-        const updatedDoc = await TotalDoneRate.findByIdAndUpdate(
-          artworkId,
-          {
-            $set: {
-              totalStartCount: totalStartCount,
-              totalDoneCount: totalDoneCount,
-              totalTipCount: totalTipCount, // 写入新的累计道具使用次数
-              completionRate: newCompletionRate,
+
+    console.log(
+      `[DoneRate Cron] 开始通过 bulkWrite 更新本地 TotalDoneRate 表。总计 ${aggregatedTotals.length} 个作品，batchSize=${TOTAL_DONE_RATE_BULK_SIZE}。`
+    );
+
+    for (let i = 0; i < aggregatedTotals.length; i += TOTAL_DONE_RATE_BULK_SIZE) {
+      const chunk = aggregatedTotals.slice(i, i + TOTAL_DONE_RATE_BULK_SIZE);
+      const ops = chunk.map((totalData) => {
+        const artworkId = totalData._id;
+        const totalStartCount = Number(totalData.totalStartCount) || 0;
+        const totalDoneCount = Number(totalData.totalDoneCount) || 0;
+        const totalTipCount = Number(totalData.totalTipCount) || 0;
+        const completionRate = totalStartCount > 0 ? (totalDoneCount / totalStartCount) * 100 : 0;
+
+        return {
+          updateOne: {
+            filter: { _id: artworkId },
+            update: {
+              $set: {
+                totalStartCount,
+                totalDoneCount,
+                totalTipCount,
+                completionRate,
+              },
             },
+            upsert: true,
           },
-          { new: true, upsert: true } // upsert: true 表示如果文档不存在则创建
-        );
-
-        if (updatedDoc) {
-          updatedTotalDoneRateCount++;
-        }
-
-        // 进度日志
-        if ((i + 1) % 50 === 0 || i === aggregatedTotals.length - 1) {
-          const elapsed = new Date().getTime() - updateStartTime;
-          console.log(`[DoneRate Cron] 进度: 已更新 ${i + 1}/${aggregatedTotals.length} 条本地 TotalDoneRate 记录, 当前耗时: ${elapsed}ms`);
-        }
-      } catch (totalUpdateError) {
-        console.error(`[DoneRate Cron] Error updating TotalDoneRate document for artwork ID ${artworkId}:`, totalUpdateError);
-      }
+        };
+      });
+
+      await TotalDoneRate.bulkWrite(ops, { ordered: false });
+      updatedTotalDoneRateCount += chunk.length;
+
+      const processed = Math.min(i + TOTAL_DONE_RATE_BULK_SIZE, aggregatedTotals.length);
+      const elapsed = Date.now() - totalUpdateStartedAt;
+      console.log(`[DoneRate Cron] 进度: 已处理 ${processed}/${aggregatedTotals.length} 条 TotalDoneRate，累计耗时 ${elapsed}ms`);
     }
 
-    const updateTimeTaken = new Date().getTime() - updateStartTime;
-    console.log(`[DoneRate Cron] 本地 TotalDoneRate 表更新完成。总计更新 ${updatedTotalDoneRateCount} 条记录，总耗时 ${updateTimeTaken}ms。`);
+    const totalUpdateElapsedMs = Date.now() - totalUpdateStartedAt;
+    console.log(
+      `[DoneRate Cron] 本地 TotalDoneRate 表更新完成。处理记录数 ${aggregatedTotals.length}，写入阶段耗时 ${totalUpdateElapsedMs}ms。`
+    );
 
     const summary = `[DoneRate Cron] Daily done-rate calculation for ${yesterdayYYYYMMDD} completed. Total DoneRate processed: ${totalProcessedArtworks}. Created DoneRate: ${createdRecordsCount}, Updated DoneRate: ${updatedRecordsCount}. Updated TotalDoneRate records: ${updatedTotalDoneRateCount}.`;
     console.log(summary);