|
@@ -2,6 +2,7 @@
|
|
|
Object.defineProperty(exports, "__esModule", { value: true });
|
|
Object.defineProperty(exports, "__esModule", { value: true });
|
|
|
exports.MessageRecordService = void 0;
|
|
exports.MessageRecordService = void 0;
|
|
|
const messageRecordModel_1 = require("../models/messageRecordModel");
|
|
const messageRecordModel_1 = require("../models/messageRecordModel");
|
|
|
|
|
+const messageStatsDailyUidModel_1 = require("../models/messageStatsDailyUidModel");
|
|
|
const clients_1 = require("./clients");
|
|
const clients_1 = require("./clients");
|
|
|
class MessageRecordService {
|
|
class MessageRecordService {
|
|
|
logPerf(endpoint, stage, durationMs, extra = {}) {
|
|
logPerf(endpoint, stage, durationMs, extra = {}) {
|
|
@@ -21,6 +22,14 @@ class MessageRecordService {
|
|
|
}
|
|
}
|
|
|
return 1;
|
|
return 1;
|
|
|
}
|
|
}
|
|
|
|
|
+ formatDateKeyInTimezone(date) {
|
|
|
|
|
+ return new Intl.DateTimeFormat("en-CA", {
|
|
|
|
|
+ timeZone: MessageRecordService.TIMEZONE,
|
|
|
|
|
+ year: "numeric",
|
|
|
|
|
+ month: "2-digit",
|
|
|
|
|
+ day: "2-digit",
|
|
|
|
|
+ }).format(date);
|
|
|
|
|
+ }
|
|
|
/**
|
|
/**
|
|
|
* 创建一条新的消息推送记录
|
|
* 创建一条新的消息推送记录
|
|
|
* @param recordData 消息记录数据
|
|
* @param recordData 消息记录数据
|
|
@@ -103,8 +112,22 @@ class MessageRecordService {
|
|
|
console.log(`[MessageStatsCache] hit key=${cacheKey}`);
|
|
console.log(`[MessageStatsCache] hit key=${cacheKey}`);
|
|
|
return cached;
|
|
return cached;
|
|
|
}
|
|
}
|
|
|
- const matchConditions = this.buildMatchConditions(startDate, endDate, strategyName);
|
|
|
|
|
- const result = await this.getStatisticsByGroup(matchConditions, []);
|
|
|
|
|
|
|
+ let result = null;
|
|
|
|
|
+ if (MessageRecordService.PREAGG_ENABLED && startDate && endDate) {
|
|
|
|
|
+ const preAggStartedAt = Date.now();
|
|
|
|
|
+ result = await this.getOverallFromPreAgg(startDate, endDate, strategyName);
|
|
|
|
|
+ if (result) {
|
|
|
|
|
+ this.logPerf("overall", "preagg", Date.now() - preAggStartedAt, {
|
|
|
|
|
+ startDate: startDate.toISOString(),
|
|
|
|
|
+ endDate: endDate.toISOString(),
|
|
|
|
|
+ hasStrategy: Boolean(strategyName),
|
|
|
|
|
+ });
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ if (!result) {
|
|
|
|
|
+ const matchConditions = this.buildMatchConditions(startDate, endDate, strategyName);
|
|
|
|
|
+ result = await this.getStatisticsByGroup(matchConditions, []);
|
|
|
|
|
+ }
|
|
|
await this.setCache(cacheKey, result);
|
|
await this.setCache(cacheKey, result);
|
|
|
console.log(`[MessageStatsCache] miss key=${cacheKey}`);
|
|
console.log(`[MessageStatsCache] miss key=${cacheKey}`);
|
|
|
return result;
|
|
return result;
|
|
@@ -133,9 +156,26 @@ class MessageRecordService {
|
|
|
console.log(`[MessageStatsCache] hit key=${cacheKey}`);
|
|
console.log(`[MessageStatsCache] hit key=${cacheKey}`);
|
|
|
return cached;
|
|
return cached;
|
|
|
}
|
|
}
|
|
|
- const matchConditions = this.buildMatchConditions(startDate, endDate, strategyName);
|
|
|
|
|
- const groupFields = ["strategyId", "strategyName"];
|
|
|
|
|
- const result = await this.getStatisticsByGroup(matchConditions, groupFields, { clickThroughRate: -1 }, page, limit);
|
|
|
|
|
|
|
+ let result = [];
|
|
|
|
|
+ if (MessageRecordService.PREAGG_ENABLED && startDate && endDate) {
|
|
|
|
|
+ const preAggStartedAt = Date.now();
|
|
|
|
|
+ result = await this.getByStrategyFromPreAgg(startDate, endDate, strategyName, page, limit);
|
|
|
|
|
+ if (Array.isArray(result) && result.length > 0) {
|
|
|
|
|
+ this.logPerf("by-strategy", "preagg", Date.now() - preAggStartedAt, {
|
|
|
|
|
+ startDate: startDate.toISOString(),
|
|
|
|
|
+ endDate: endDate.toISOString(),
|
|
|
|
|
+ hasStrategy: Boolean(strategyName),
|
|
|
|
|
+ page,
|
|
|
|
|
+ limit,
|
|
|
|
|
+ resultRows: result.length,
|
|
|
|
|
+ });
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ if (!Array.isArray(result) || result.length === 0) {
|
|
|
|
|
+ const matchConditions = this.buildMatchConditions(startDate, endDate, strategyName);
|
|
|
|
|
+ const groupFields = ["strategyId", "strategyName"];
|
|
|
|
|
+ result = await this.getStatisticsByGroup(matchConditions, groupFields, { clickThroughRate: -1 }, page, limit);
|
|
|
|
|
+ }
|
|
|
await this.setCache(cacheKey, result);
|
|
await this.setCache(cacheKey, result);
|
|
|
console.log(`[MessageStatsCache] miss key=${cacheKey}`);
|
|
console.log(`[MessageStatsCache] miss key=${cacheKey}`);
|
|
|
return result;
|
|
return result;
|
|
@@ -222,7 +262,22 @@ class MessageRecordService {
|
|
|
console.log(`[MessageStatsCache] hit key=${cacheKey}`);
|
|
console.log(`[MessageStatsCache] hit key=${cacheKey}`);
|
|
|
return cached;
|
|
return cached;
|
|
|
}
|
|
}
|
|
|
- const result = await this.getDailyTrendsByDimensions(startDate, endDate, strategyName, []);
|
|
|
|
|
|
|
+ let result = [];
|
|
|
|
|
+ if (MessageRecordService.PREAGG_ENABLED && startDate && endDate) {
|
|
|
|
|
+ const preAggStartedAt = Date.now();
|
|
|
|
|
+ result = await this.getDailyTrendsFromPreAgg(startDate, endDate, strategyName);
|
|
|
|
|
+ if (Array.isArray(result) && result.length > 0) {
|
|
|
|
|
+ this.logPerf("daily-trends", "preagg", Date.now() - preAggStartedAt, {
|
|
|
|
|
+ startDate: startDate.toISOString(),
|
|
|
|
|
+ endDate: endDate.toISOString(),
|
|
|
|
|
+ hasStrategy: Boolean(strategyName),
|
|
|
|
|
+ resultRows: result.length,
|
|
|
|
|
+ });
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ if (!Array.isArray(result) || result.length === 0) {
|
|
|
|
|
+ result = await this.getDailyTrendsByDimensions(startDate, endDate, strategyName, []);
|
|
|
|
|
+ }
|
|
|
await this.setCache(cacheKey, result);
|
|
await this.setCache(cacheKey, result);
|
|
|
console.log(`[MessageStatsCache] miss key=${cacheKey}`);
|
|
console.log(`[MessageStatsCache] miss key=${cacheKey}`);
|
|
|
return result;
|
|
return result;
|
|
@@ -274,9 +329,34 @@ class MessageRecordService {
|
|
|
pagination: { page: safePage, limit: safeLimit },
|
|
pagination: { page: safePage, limit: safeLimit },
|
|
|
};
|
|
};
|
|
|
}
|
|
}
|
|
|
- // 任意 cache miss → 单次 $facet 查询替代 3 次独立聚合
|
|
|
|
|
- const matchConditions = this.buildMatchConditions(startDate, endDate, strategyName);
|
|
|
|
|
- const { overall, strategyStats, dailyTrends } = await this.runSummaryFacetQuery(matchConditions, safePage, safeLimit);
|
|
|
|
|
|
|
+ let overall = null;
|
|
|
|
|
+ let strategyStats = [];
|
|
|
|
|
+ let dailyTrends = [];
|
|
|
|
|
+ // 任意 cache miss:优先读取预聚合,预聚合不可用时回退实时 facet
|
|
|
|
|
+ if (MessageRecordService.PREAGG_ENABLED && startDate && endDate) {
|
|
|
|
|
+ const preAggResult = await this.runSummaryFromPreAgg(startDate, endDate, strategyName, safePage, safeLimit);
|
|
|
|
|
+ overall = preAggResult.overall;
|
|
|
|
|
+ strategyStats = preAggResult.strategyStats;
|
|
|
|
|
+ dailyTrends = preAggResult.dailyTrends;
|
|
|
|
|
+ this.logPerf("summary", "preagg-query", Date.now() - summaryStartedAt, {
|
|
|
|
|
+ overallRows: this.resolveResultRows(overall),
|
|
|
|
|
+ strategyRows: this.resolveResultRows(strategyStats),
|
|
|
|
|
+ dailyTrendRows: this.resolveResultRows(dailyTrends),
|
|
|
|
|
+ });
|
|
|
|
|
+ }
|
|
|
|
|
+ if (!overall || strategyStats.length === 0 || dailyTrends.length === 0) {
|
|
|
|
|
+ const matchConditions = this.buildMatchConditions(startDate, endDate, strategyName);
|
|
|
|
|
+ const facetResult = await this.runSummaryFacetQuery(matchConditions, safePage, safeLimit);
|
|
|
|
|
+ if (!overall) {
|
|
|
|
|
+ overall = facetResult.overall;
|
|
|
|
|
+ }
|
|
|
|
|
+ if (strategyStats.length === 0) {
|
|
|
|
|
+ strategyStats = facetResult.strategyStats;
|
|
|
|
|
+ }
|
|
|
|
|
+ if (dailyTrends.length === 0) {
|
|
|
|
|
+ dailyTrends = facetResult.dailyTrends;
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
// 回填各子查询缓存(未命中的才写入)
|
|
// 回填各子查询缓存(未命中的才写入)
|
|
|
await Promise.all([
|
|
await Promise.all([
|
|
|
cachedOverall ? Promise.resolve() : this.setCache(overallCacheKey, overall),
|
|
cachedOverall ? Promise.resolve() : this.setCache(overallCacheKey, overall),
|
|
@@ -537,6 +617,170 @@ class MessageRecordService {
|
|
|
},
|
|
},
|
|
|
};
|
|
};
|
|
|
}
|
|
}
|
|
|
|
|
+ buildPreAggMatch(startDate, endDate, strategyName) {
|
|
|
|
|
+ const startKey = this.formatDateKeyInTimezone(startDate);
|
|
|
|
|
+ const endKey = this.formatDateKeyInTimezone(endDate);
|
|
|
|
|
+ const match = {
|
|
|
|
|
+ dateKey: {
|
|
|
|
|
+ $gte: startKey,
|
|
|
|
|
+ $lte: endKey,
|
|
|
|
|
+ },
|
|
|
|
|
+ };
|
|
|
|
|
+ if (strategyName) {
|
|
|
|
|
+ match.strategyName = strategyName;
|
|
|
|
|
+ }
|
|
|
|
|
+ return match;
|
|
|
|
|
+ }
|
|
|
|
|
+ async getOverallFromPreAgg(startDate, endDate, strategyName) {
|
|
|
|
|
+ const pipeline = [
|
|
|
|
|
+ { $match: this.buildPreAggMatch(startDate, endDate, strategyName) },
|
|
|
|
|
+ {
|
|
|
|
|
+ $group: {
|
|
|
|
|
+ _id: { status: "$status", inforeground: "$inforeground", uid: "$uid" },
|
|
|
|
|
+ count: { $sum: "$msgCount" },
|
|
|
|
|
+ },
|
|
|
|
|
+ },
|
|
|
|
|
+ {
|
|
|
|
|
+ $group: {
|
|
|
|
|
+ _id: { status: "$_id.status", inforeground: "$_id.inforeground" },
|
|
|
|
|
+ count: { $sum: "$count" },
|
|
|
|
|
+ uniqueUsers: { $sum: 1 },
|
|
|
|
|
+ },
|
|
|
|
|
+ },
|
|
|
|
|
+ {
|
|
|
|
|
+ $group: {
|
|
|
|
|
+ _id: null,
|
|
|
|
|
+ totalRecords: { $sum: "$count" },
|
|
|
|
|
+ ...this.getStatusAggregationFields(),
|
|
|
|
|
+ },
|
|
|
|
|
+ },
|
|
|
|
|
+ { $project: this.getStatisticsProjectFields([]) },
|
|
|
|
|
+ ];
|
|
|
|
|
+ const [result] = await messageStatsDailyUidModel_1.MessageStatsDailyUid.aggregate(pipeline).allowDiskUse(true);
|
|
|
|
|
+ return result || null;
|
|
|
|
|
+ }
|
|
|
|
|
+ async getByStrategyFromPreAgg(startDate, endDate, strategyName, page, limit) {
|
|
|
|
|
+ const safePage = Math.max(1, Math.floor(page || MessageRecordService.DEFAULT_STATS_PAGE));
|
|
|
|
|
+ const safeLimit = Math.min(MessageRecordService.MAX_STATS_LIMIT, Math.max(1, Math.floor(limit || MessageRecordService.DEFAULT_STATS_LIMIT)));
|
|
|
|
|
+ const pipeline = [
|
|
|
|
|
+ { $match: this.buildPreAggMatch(startDate, endDate, strategyName) },
|
|
|
|
|
+ {
|
|
|
|
|
+ $group: {
|
|
|
|
|
+ _id: {
|
|
|
|
|
+ strategyId: "$strategyId",
|
|
|
|
|
+ strategyName: "$strategyName",
|
|
|
|
|
+ status: "$status",
|
|
|
|
|
+ inforeground: "$inforeground",
|
|
|
|
|
+ uid: "$uid",
|
|
|
|
|
+ },
|
|
|
|
|
+ count: { $sum: "$msgCount" },
|
|
|
|
|
+ },
|
|
|
|
|
+ },
|
|
|
|
|
+ {
|
|
|
|
|
+ $group: {
|
|
|
|
|
+ _id: {
|
|
|
|
|
+ strategyId: "$_id.strategyId",
|
|
|
|
|
+ strategyName: "$_id.strategyName",
|
|
|
|
|
+ status: "$_id.status",
|
|
|
|
|
+ inforeground: "$_id.inforeground",
|
|
|
|
|
+ },
|
|
|
|
|
+ count: { $sum: "$count" },
|
|
|
|
|
+ uniqueUsers: { $sum: 1 },
|
|
|
|
|
+ },
|
|
|
|
|
+ },
|
|
|
|
|
+ {
|
|
|
|
|
+ $group: {
|
|
|
|
|
+ _id: { strategyId: "$_id.strategyId", strategyName: "$_id.strategyName" },
|
|
|
|
|
+ strategyId: { $first: "$_id.strategyId" },
|
|
|
|
|
+ strategyName: { $first: "$_id.strategyName" },
|
|
|
|
|
+ totalRecords: { $sum: "$count" },
|
|
|
|
|
+ ...this.getStatusAggregationFields(),
|
|
|
|
|
+ },
|
|
|
|
|
+ },
|
|
|
|
|
+ { $project: this.getStatisticsProjectFields(["strategyId", "strategyName"]) },
|
|
|
|
|
+ { $sort: { clickThroughRate: -1 } },
|
|
|
|
|
+ { $skip: (safePage - 1) * safeLimit },
|
|
|
|
|
+ { $limit: safeLimit },
|
|
|
|
|
+ ];
|
|
|
|
|
+ return messageStatsDailyUidModel_1.MessageStatsDailyUid.aggregate(pipeline).allowDiskUse(true);
|
|
|
|
|
+ }
|
|
|
|
|
+ async getDailyTrendsFromPreAgg(startDate, endDate, strategyName) {
|
|
|
|
|
+ const pipeline = [
|
|
|
|
|
+ { $match: this.buildPreAggMatch(startDate, endDate, strategyName) },
|
|
|
|
|
+ {
|
|
|
|
|
+ $group: {
|
|
|
|
|
+ _id: {
|
|
|
|
|
+ date: "$date",
|
|
|
|
|
+ status: "$status",
|
|
|
|
|
+ inforeground: "$inforeground",
|
|
|
|
|
+ uid: "$uid",
|
|
|
|
|
+ },
|
|
|
|
|
+ count: { $sum: "$msgCount" },
|
|
|
|
|
+ },
|
|
|
|
|
+ },
|
|
|
|
|
+ {
|
|
|
|
|
+ $group: {
|
|
|
|
|
+ _id: { date: "$_id.date", status: "$_id.status", inforeground: "$_id.inforeground" },
|
|
|
|
|
+ count: { $sum: "$count" },
|
|
|
|
|
+ uniqueUsers: { $sum: 1 },
|
|
|
|
|
+ },
|
|
|
|
|
+ },
|
|
|
|
|
+ {
|
|
|
|
|
+ $group: {
|
|
|
|
|
+ _id: "$_id.date",
|
|
|
|
|
+ date: { $first: "$_id.date" },
|
|
|
|
|
+ totalRecords: { $sum: "$count" },
|
|
|
|
|
+ ...this.getStatusAggregationFields(),
|
|
|
|
|
+ },
|
|
|
|
|
+ },
|
|
|
|
|
+ {
|
|
|
|
|
+ $project: {
|
|
|
|
|
+ _id: 0,
|
|
|
|
|
+ date: "$date",
|
|
|
|
|
+ totalRecords: "$totalRecords",
|
|
|
|
|
+ sent: "$sent",
|
|
|
|
|
+ delivered: "$delivered",
|
|
|
|
|
+ opened: "$opened",
|
|
|
|
|
+ failed: "$failed",
|
|
|
|
|
+ displayCount: "$displayCount",
|
|
|
|
|
+ displayedUsers: "$displayedUsers",
|
|
|
|
|
+ openedUsers: "$openedUsers",
|
|
|
|
|
+ sentSuccessRate: {
|
|
|
|
|
+ $cond: [{ $eq: ["$totalRecords", 0] }, 0, { $divide: ["$sent", "$totalRecords"] }],
|
|
|
|
|
+ },
|
|
|
|
|
+ deliveredRate: {
|
|
|
|
|
+ $cond: [{ $eq: ["$sent", 0] }, 0, { $divide: ["$delivered", "$sent"] }],
|
|
|
|
|
+ },
|
|
|
|
|
+ displayRate: {
|
|
|
|
|
+ $cond: [{ $eq: ["$delivered", 0] }, 0, { $divide: ["$displayCount", "$delivered"] }],
|
|
|
|
|
+ },
|
|
|
|
|
+ clickThroughRate: {
|
|
|
|
|
+ $cond: [{ $eq: ["$displayCount", 0] }, 0, { $divide: ["$opened", "$displayCount"] }],
|
|
|
|
|
+ },
|
|
|
|
|
+ tokenInvalidationRate: {
|
|
|
|
|
+ $cond: [{ $eq: ["$totalRecords", 0] }, 0, { $divide: ["$failed", "$totalRecords"] }],
|
|
|
|
|
+ },
|
|
|
|
|
+ actualClickThroughRate: {
|
|
|
|
|
+ $cond: [{ $eq: ["$displayedUsers", 0] }, 0, { $divide: ["$openedUsers", "$displayedUsers"] }],
|
|
|
|
|
+ },
|
|
|
|
|
+ },
|
|
|
|
|
+ },
|
|
|
|
|
+ { $sort: { date: -1 } },
|
|
|
|
|
+ ];
|
|
|
|
|
+ return messageStatsDailyUidModel_1.MessageStatsDailyUid.aggregate(pipeline).allowDiskUse(true);
|
|
|
|
|
+ }
|
|
|
|
|
+ async runSummaryFromPreAgg(startDate, endDate, strategyName, page, limit) {
|
|
|
|
|
+ const [overall, strategyStats, dailyTrends] = await Promise.all([
|
|
|
|
|
+ this.getOverallFromPreAgg(startDate, endDate, strategyName),
|
|
|
|
|
+ this.getByStrategyFromPreAgg(startDate, endDate, strategyName, page, limit),
|
|
|
|
|
+ this.getDailyTrendsFromPreAgg(startDate, endDate, strategyName),
|
|
|
|
|
+ ]);
|
|
|
|
|
+ return {
|
|
|
|
|
+ overall,
|
|
|
|
|
+ strategyStats: strategyStats || [],
|
|
|
|
|
+ dailyTrends: dailyTrends || [],
|
|
|
|
|
+ };
|
|
|
|
|
+ }
|
|
|
/**
|
|
/**
|
|
|
* summary 首屏接口专用:单次 $facet 查询同时计算 overall / byStrategy / dailyTrends,
|
|
* summary 首屏接口专用:单次 $facet 查询同时计算 overall / byStrategy / dailyTrends,
|
|
|
* 替代 3 次独立聚合,减少对同一时间窗口数据的重复扫描。
|
|
* 替代 3 次独立聚合,减少对同一时间窗口数据的重复扫描。
|
|
@@ -928,3 +1172,4 @@ MessageRecordService.DEFAULT_STATS_PAGE = 1;
|
|
|
MessageRecordService.DEFAULT_STATS_LIMIT = 50;
|
|
MessageRecordService.DEFAULT_STATS_LIMIT = 50;
|
|
|
MessageRecordService.MAX_STATS_LIMIT = 200;
|
|
MessageRecordService.MAX_STATS_LIMIT = 200;
|
|
|
MessageRecordService.STATS_CACHE_TTL_SECONDS = 300;
|
|
MessageRecordService.STATS_CACHE_TTL_SECONDS = 300;
|
|
|
|
|
+MessageRecordService.PREAGG_ENABLED = process.env.MESSAGE_STATS_PREAGG_ENABLED === "1";
|