87 lines
3.1 KiB
TypeScript
87 lines
3.1 KiB
TypeScript
import { Injectable, Logger, Inject } from '@nestjs/common';
|
|
import { Cron, CronExpression } from '@nestjs/schedule';
|
|
import { CommandBus } from '@nestjs/cqrs';
|
|
import { subDays } from 'date-fns';
|
|
import { CalculateDauCommand } from '../commands/calculate-dau/calculate-dau.command';
|
|
import { PresenceRedisRepository } from '../../infrastructure/redis/presence-redis.repository';
|
|
import { OnlineDetectionService } from '../../domain/services/online-detection.service';
|
|
import { OnlineSnapshot } from '../../domain/entities/online-snapshot.entity';
|
|
import { IOnlineSnapshotRepository, ONLINE_SNAPSHOT_REPOSITORY } from '../../domain/repositories/online-snapshot.repository.interface';
|
|
|
|
@Injectable()
|
|
export class AnalyticsScheduler {
|
|
private readonly logger = new Logger(AnalyticsScheduler.name);
|
|
|
|
constructor(
|
|
private readonly commandBus: CommandBus,
|
|
private readonly presenceRedisRepository: PresenceRedisRepository,
|
|
private readonly onlineDetectionService: OnlineDetectionService,
|
|
@Inject(ONLINE_SNAPSHOT_REPOSITORY)
|
|
private readonly snapshotRepository: IOnlineSnapshotRepository,
|
|
) {}
|
|
|
|
/**
|
|
* 每分钟记录在线人数快照
|
|
*/
|
|
@Cron(CronExpression.EVERY_MINUTE)
|
|
async recordOnlineSnapshot(): Promise<void> {
|
|
try {
|
|
const now = new Date();
|
|
const threshold = this.onlineDetectionService.getOnlineThreshold(now);
|
|
const count = await this.presenceRedisRepository.countOnlineUsers(threshold);
|
|
|
|
const snapshot = OnlineSnapshot.create({
|
|
ts: now,
|
|
onlineCount: count,
|
|
windowSeconds: this.onlineDetectionService.getWindowSeconds(),
|
|
});
|
|
|
|
await this.snapshotRepository.insert(snapshot);
|
|
this.logger.debug(`Online snapshot recorded: ${count} users`);
|
|
} catch (error) {
|
|
this.logger.error('Failed to record online snapshot', error);
|
|
}
|
|
}
|
|
|
|
/**
|
|
* 每小时清理过期在线数据
|
|
*/
|
|
@Cron(CronExpression.EVERY_HOUR)
|
|
async cleanupExpiredPresence(): Promise<void> {
|
|
try {
|
|
const threshold = Math.floor(Date.now() / 1000) - 86400; // 24小时前
|
|
await this.presenceRedisRepository.removeExpiredUsers(threshold);
|
|
this.logger.log('Expired presence data cleaned up');
|
|
} catch (error) {
|
|
this.logger.error('Failed to cleanup expired presence', error);
|
|
}
|
|
}
|
|
|
|
/**
|
|
* 每天凌晨 1:00 计算前一天 DAU (Asia/Shanghai)
|
|
*/
|
|
@Cron('0 0 1 * * *', { timeZone: 'Asia/Shanghai' })
|
|
async calculateYesterdayDau(): Promise<void> {
|
|
try {
|
|
const yesterday = subDays(new Date(), 1);
|
|
await this.commandBus.execute(new CalculateDauCommand(yesterday));
|
|
this.logger.log('Yesterday DAU calculated');
|
|
} catch (error) {
|
|
this.logger.error('Failed to calculate yesterday DAU', error);
|
|
}
|
|
}
|
|
|
|
/**
|
|
* 每小时滚动计算当天 DAU (用于实时看板)
|
|
*/
|
|
@Cron(CronExpression.EVERY_HOUR)
|
|
async calculateTodayDauRolling(): Promise<void> {
|
|
try {
|
|
await this.commandBus.execute(new CalculateDauCommand(new Date()));
|
|
this.logger.debug('Today DAU rolling calculation completed');
|
|
} catch (error) {
|
|
this.logger.error('Failed to calculate today DAU', error);
|
|
}
|
|
}
|
|
}
|