feat(ip-report): add IP reporting feature with hourly and daily aggregation capabilities

.gitignore

@@ -75,3 +75,4 @@ timestamp-audit.json
 apps/box-app-api/src/feature/homepage/README.md
 libs/core/media-manager/CONTRACT.md
 libs/core/media-manager/CONTRACT.md
+scripts/*

apps/box-mgnt-api/src/config/env.validation.ts

@@ -48,6 +48,31 @@ class EnvironmentVariables {
   @IsOptional()
   MGNT_PORT: number = 3000;
 
+  @IsInt()
+  @Min(1)
+  @IsOptional()
+  IP_REPORT_MAX_WINDOW_HOURS: number = 48; // sliding window limit for the IP report query
+
+  @IsInt()
+  @Min(1)
+  @IsOptional()
+  IP_REPORT_RAW_RETENTION_DAYS: number = 30; // raw data retention in days for IP report cutoffs
+
+  @IsInt()
+  @Min(1)
+  @IsOptional()
+  IP_REPORT_MAX_EVENTS: number = 200000; // guardrail to keep report queries bounded
+
+  @IsInt()
+  @Min(1)
+  @IsOptional()
+  IP_REPORT_MAX_ROWS: number = 50000; // guardrail for aggregated row count
+
+  @IsInt()
+  @Min(0)
+  @IsOptional()
+  IP_REPORT_LOGIN_LOOKBACK_SEC: number = 3600; // how far behind fromSec to look for login records
+
   // ===== Redis Config =====
 
   @IsString()

apps/box-mgnt-api/src/mgnt-backend/feature/feature.module.ts

@@ -12,6 +12,7 @@ import { VideoMediaModule } from './video-media/video-media.module';
 import { HealthModule } from './health/health.module';
 import { ProviderVideoSyncModule } from './provider-video-sync/provider-video-sync.module';
 import { RedisInspectorModule } from './redis-inspector/redis-inspector.module';
+import { IpReportModule } from './ip-report/ip-report.module';
 
 @Module({
   imports: [
@@ -27,6 +28,7 @@ import { RedisInspectorModule } from './redis-inspector/redis-inspector.module';
     HealthModule,
     RedisInspectorModule,
     ProviderVideoSyncModule,
+    IpReportModule,
   ],
 })
 export class FeatureModule {}

apps/box-mgnt-api/src/mgnt-backend/feature/ip-report/dto/ip-level-hourly-report-query.dto.ts

@@ -0,0 +1,102 @@
+import { Transform } from 'class-transformer';
+import { ApiProperty, ApiPropertyOptional } from '@nestjs/swagger';
+import {
+  IsInt,
+  Min,
+  Validate,
+  IsOptional,
+  IsString,
+  IsMongoId,
+  IsIP,
+} from 'class-validator';
+import {
+  IpReportHourAlignedValidator,
+  IpReportWindowValidator,
+} from '../validators/ip-report.validators';
+
+const parseEpochValue = (value: unknown) => {
+  if (typeof value === 'string') {
+    const trimmed = value.trim();
+    if (trimmed === '') {
+      return value;
+    }
+    const parsed = Number(trimmed);
+    return Number.isNaN(parsed) ? value : parsed;
+  }
+
+  return value;
+};
+
+const trimString = (value: unknown) => {
+  if (typeof value === 'string') {
+    const trimmed = value.trim();
+    return trimmed === '' ? undefined : trimmed;
+  }
+  return value;
+};
+
+export class IpLevelHourlyReportQueryDto {
+  @ApiProperty({
+    description:
+      'Inclusive start epoch seconds for the hourly report (strings allowed for bigint parsing, must be hour-aligned).',
+    example: 1693449600,
+  })
+  @Transform(({ value }) => parseEpochValue(value))
+  @IsInt()
+  @Min(0)
+  @Validate(IpReportHourAlignedValidator)
+  fromSec: number;
+
+  @ApiProperty({
+    description:
+      'Exclusive end epoch seconds (half-open range) for the hourly report (string acceptable, hour-aligned).',
+    example: 1693453200,
+  })
+  @Transform(({ value }) => parseEpochValue(value))
+  @IsInt()
+  @Min(0)
+  @Validate(IpReportHourAlignedValidator)
+  @Validate(IpReportWindowValidator)
+  toSec: number;
+
+  @ApiPropertyOptional({
+    description:
+      'Channel identifier (matches both AdClickEvents.channelId and UserLoginHistory.channelId).',
+  })
+  @IsOptional()
+  @IsString()
+  @Transform(({ value }) => trimString(value))
+  channelId?: string;
+
+  @ApiPropertyOptional({
+    description: 'Filters by adsId (AdClickEvents.adsId).',
+  })
+  @IsOptional()
+  @IsMongoId()
+  @Transform(({ value }) => trimString(value))
+  adsId?: string;
+
+  @ApiPropertyOptional({
+    description: 'Filter by adType (AdClickEvents.adType).',
+  })
+  @IsOptional()
+  @IsString()
+  @Transform(({ value }) => trimString(value))
+  adType?: string;
+
+  @ApiPropertyOptional({
+    description: 'Filter by login IP (maps to UserLoginHistory.ip).',
+  })
+  @IsOptional()
+  @IsIP()
+  @Transform(({ value }) => trimString(value))
+  loginIp?: string;
+
+  @ApiPropertyOptional({
+    description: 'Filter by click IP (maps to AdClickEvents.ip).',
+  })
+  @IsOptional()
+  @IsIP()
+  @Transform(({ value }) => trimString(value))
+  clickIp?: string;
+}

apps/box-mgnt-api/src/mgnt-backend/feature/ip-report/ip-report-time.helper.ts

@@ -0,0 +1,8 @@
+const TZ_OFFSET_SEC = 8 * 3600;
+
+export function getGmt8HourStartUtcSec(epochSec: number | bigint): number {
+  const numeric = typeof epochSec === 'bigint' ? Number(epochSec) : epochSec;
+  const shifted = numeric + TZ_OFFSET_SEC;
+  const hourStartShifted = Math.floor(shifted / 3600) * 3600;
+  return hourStartShifted - TZ_OFFSET_SEC;
+}

apps/box-mgnt-api/src/mgnt-backend/feature/ip-report/ip-report.config.ts

@@ -0,0 +1,11 @@
+import { ConfigType, registerAs } from '@nestjs/config';
+
+export const ipReportConfigFactory = registerAs('ipReport', () => ({
+  maxWindowHours: Number(process.env.IP_REPORT_MAX_WINDOW_HOURS ?? 48),
+  rawRetentionDays: Number(process.env.IP_REPORT_RAW_RETENTION_DAYS ?? 30),
+  maxEvents: Number(process.env.IP_REPORT_MAX_EVENTS ?? 200000),
+  maxRows: Number(process.env.IP_REPORT_MAX_ROWS ?? 50000),
+  loginLookbackSec: Number(process.env.IP_REPORT_LOGIN_LOOKBACK_SEC ?? 3600),
+}));
+
+export type IpReportConfig = ConfigType<typeof ipReportConfigFactory>;

apps/box-mgnt-api/src/mgnt-backend/feature/ip-report/ip-report.controller.ts

@@ -0,0 +1,35 @@
+import { Body, Controller, Post } from '@nestjs/common';
+import { ApiOperation, ApiResponse, ApiTags } from '@nestjs/swagger';
+import { IpLevelHourlyReportQueryDto } from './dto/ip-level-hourly-report-query.dto';
+import { IpReportService } from './ip-report.service';
+
+@ApiTags('IP Reporting')
+@Controller('reports/ip-hourly')
+export class IpReportController {
+  constructor(private readonly service: IpReportService) {}
+
+  @Post()
+  @ApiOperation({ summary: 'Generate hourly IP-level report data' })
+  @ApiResponse({
+    status: 200,
+    description:
+      'Placeholder response for IP report (rows are empty until aggregation is implemented).',
+    schema: {
+      type: 'object',
+      properties: {
+        rows: { type: 'array', items: { type: 'object' } },
+        meta: {
+          type: 'object',
+          properties: {
+            fromSec: { type: 'integer', example: 1693449600 },
+            toSec: { type: 'integer', example: 1693453200 },
+            windowHours: { type: 'number', example: 1 },
+          },
+        },
+      },
+    },
+  })
+  getReport(@Body() dto: IpLevelHourlyReportQueryDto) {
+    return this.service.getIpLevelHourlyReport(dto);
+  }
+}

apps/box-mgnt-api/src/mgnt-backend/feature/ip-report/ip-report.module.ts

@@ -0,0 +1,22 @@
+import { Module } from '@nestjs/common';
+import { ConfigModule } from '@nestjs/config';
+import { PrismaModule } from '@box/db/prisma/prisma.module';
+import { ipReportConfigFactory } from './ip-report.config';
+import {
+  IpReportHourAlignedValidator,
+  IpReportWindowValidator,
+} from './validators/ip-report.validators';
+import { IpReportController } from './ip-report.controller';
+import { IpReportService } from './ip-report.service';
+
+@Module({
+  imports: [ConfigModule.forFeature(ipReportConfigFactory), PrismaModule],
+  controllers: [IpReportController],
+  providers: [
+    IpReportHourAlignedValidator,
+    IpReportWindowValidator,
+    IpReportService,
+  ],
+  exports: [IpReportHourAlignedValidator, IpReportWindowValidator],
+})
+export class IpReportModule {}

apps/box-mgnt-api/src/mgnt-backend/feature/ip-report/ip-report.service.ts

@@ -0,0 +1,248 @@
+import { BadRequestException, Injectable, Logger } from '@nestjs/common';
+import { Prisma } from '@prisma/mongo-stats/client';
+import { MongoStatsPrismaService } from '@box/db/prisma/mongo-stats-prisma.service';
+import { IpLevelHourlyReportQueryDto } from './dto/ip-level-hourly-report-query.dto';
+import { IpLevelHourlyReportRow } from './types/ip-report.types';
+import { getGmt8HourStartUtcSec } from './ip-report-time.helper';
+import { ConfigService } from '@nestjs/config';
+
+type LoginLookupValue = {
+  loginIp: string;
+  osCategory: 'ANDROID' | 'IOS' | 'OTHER';
+  createAt: bigint;
+};
+
+type AggregationEntry = {
+  hourSec: number;
+  channelId: string | null;
+  adType: string | null;
+  adsId: string | null;
+  loginIp: string | null;
+  clickIp: string | null;
+  osCategory: 'ANDROID' | 'IOS' | 'OTHER';
+  clickCount: bigint;
+};
+
+const OS_ANDROID = 'ANDROID';
+const OS_IOS = 'IOS';
+const OS_OTHER = 'OTHER';
+
+// eslint-disable-next-line @typescript-eslint/no-unused-vars
+type _PrismaGuard = Prisma.AdClickEventsWhereInput;
+
+@Injectable()
+export class IpReportService {
+  private readonly logger = new Logger(IpReportService.name);
+
+  // IP report must use MongoStatsPrismaService/@prisma/mongo-stats because AdClickEvents and UserLoginHistory live there.
+  constructor(
+    private readonly mongoPrisma: MongoStatsPrismaService,
+    private readonly configService: ConfigService,
+  ) {}
+
+  async getIpLevelHourlyReport(dto: IpLevelHourlyReportQueryDto) {
+    const startMs = Date.now();
+    const fromBig = BigInt(dto.fromSec);
+    const toBig = BigInt(dto.toSec);
+    const loginLookbackSec =
+      this.configService.get<number>('ipReport.loginLookbackSec') ?? 3600;
+    const maxEvents =
+      this.configService.get<number>('ipReport.maxEvents') ?? 200000;
+    const maxRows = this.configService.get<number>('ipReport.maxRows') ?? 50000;
+    const loginStartSec = Math.max(0, dto.fromSec - loginLookbackSec);
+    const loginStartBig = BigInt(loginStartSec);
+
+    const clickFilter: Prisma.AdClickEventsWhereInput = {
+      clickedAt: { gte: fromBig, lt: toBig },
+      ...(dto.channelId ? { channelId: dto.channelId } : {}),
+      ...(dto.adsId ? { adsId: dto.adsId } : {}),
+      ...(dto.adType ? { adType: dto.adType } : {}),
+      ...(dto.clickIp ? { ip: dto.clickIp } : {}),
+    };
+
+    const clickRecords = await this.mongoPrisma.adClickEvents.findMany({
+      where: clickFilter,
+      select: {
+        uid: true,
+        channelId: true,
+        adsId: true,
+        adType: true,
+        clickedAt: true,
+        ip: true,
+      },
+    });
+
+    if (clickRecords.length > maxEvents) {
+      const summary = {
+        fromSec: dto.fromSec,
+        toSec: dto.toSec,
+        channelId: dto.channelId,
+        adType: dto.adType,
+        adsId: dto.adsId,
+        clickIp: dto.clickIp,
+      };
+      this.logger.warn(
+        `IP report rejected; event count ${clickRecords.length} > ${maxEvents} filters=${JSON.stringify(
+          summary,
+        )}`,
+      );
+      throw new BadRequestException(
+        `IP report query matches ${clickRecords.length} clicks which exceeds the limit of ${maxEvents}. Narrow the window or add additional filters.`,
+      );
+    }
+
+    const loginFilter: Prisma.UserLoginHistoryWhereInput = {
+      createAt: { gte: loginStartBig, lt: toBig },
+      ...(dto.channelId ? { channelId: dto.channelId } : {}),
+      ...(dto.loginIp ? { ip: dto.loginIp } : {}),
+    };
+
+    const loginRecords = await this.mongoPrisma.userLoginHistory.findMany({
+      where: loginFilter,
+      select: {
+        uid: true,
+        createAt: true,
+        ip: true,
+        os: true,
+      },
+    });
+
+    const loginMap = new Map<string, LoginLookupValue>();
+
+    for (const record of loginRecords) {
+      if (!record.uid) {
+        continue;
+      }
+
+      const hourSec = getGmt8HourStartUtcSec(record.createAt ?? 0);
+      const key = `${record.uid}|${hourSec}`;
+      const existing = loginMap.get(key);
+
+      if (!existing || record.createAt > existing.createAt) {
+        loginMap.set(key, {
+          loginIp: record.ip ?? '',
+          osCategory: this.resolveOsCategory(record.os),
+          createAt: record.createAt ?? BigInt(0),
+        });
+      }
+    }
+
+    const aggregates = new Map<string, AggregationEntry>();
+
+    for (const click of clickRecords) {
+      if (!click.uid) {
+        continue;
+      }
+
+      const hourSec = getGmt8HourStartUtcSec(click.clickedAt ?? 0);
+      const loginKey = `${click.uid}|${hourSec}`;
+      const loginInfo = loginMap.get(loginKey);
+      const loginIpKey = loginInfo?.loginIp ?? '';
+      const osCategory = loginInfo?.osCategory ?? OS_OTHER;
+      const clickIpKey = click.ip ?? '';
+      const aggregationKey = [
+        hourSec,
+        click.channelId ?? '',
+        click.adType ?? '',
+        click.adsId ?? '',
+        loginIpKey,
+        clickIpKey,
+        osCategory,
+      ].join('|');
+
+      const entry = aggregates.get(aggregationKey);
+      if (entry) {
+        entry.clickCount += 1n;
+        continue;
+      }
+
+      aggregates.set(aggregationKey, {
+        hourSec,
+        channelId: click.channelId ?? null,
+        adType: click.adType ?? null,
+        adsId: click.adsId ?? null,
+        loginIp: loginIpKey || null,
+        clickIp: clickIpKey || null,
+        osCategory,
+        clickCount: 1n,
+      });
+    }
+
+    const rows: IpLevelHourlyReportRow[] = Array.from(aggregates.values())
+      .map((entry) => ({
+        hourSec: entry.hourSec,
+        channelId: entry.channelId,
+        adType: entry.adType,
+        adsId: entry.adsId,
+        loginIp: entry.loginIp,
+        clickIp: entry.clickIp,
+        osCategory: entry.osCategory,
+        clickCount: entry.clickCount.toString(),
+      }))
+      .sort((a, b) => {
+        if (a.hourSec !== b.hourSec) {
+          return a.hourSec - b.hourSec;
+        }
+
+        const channelA = a.channelId ?? '';
+        const channelB = b.channelId ?? '';
+        if (channelA !== channelB) {
+          return channelA.localeCompare(channelB);
+        }
+
+        const adsA = a.adsId ?? '';
+        const adsB = b.adsId ?? '';
+        return adsA.localeCompare(adsB);
+      });
+
+    if (rows.length > maxRows) {
+      const summary = {
+        fromSec: dto.fromSec,
+        toSec: dto.toSec,
+        channelId: dto.channelId,
+        adType: dto.adType,
+        adsId: dto.adsId,
+      };
+      this.logger.warn(
+        `IP report rejected; row count ${rows.length} > ${maxRows} filters=${JSON.stringify(
+          summary,
+        )}`,
+      );
+      throw new BadRequestException(
+        `IP report result would contain ${rows.length} rows which exceeds the limit of ${maxRows}. Narrow the grouping dimensions or time window.`,
+      );
+    }
+
+    const elapsedMs = Date.now() - startMs;
+    this.logger.log(
+      `IP report aggregated rows=${rows.length} clicks=${clickRecords.length} logins=${loginRecords.length} took=${elapsedMs}ms`,
+    );
+
+    return {
+      rows,
+      meta: {
+        fromSec: dto.fromSec,
+        toSec: dto.toSec,
+        windowHours: (dto.toSec - dto.fromSec) / 3600,
+      },
+    };
+  }
+
+  private resolveOsCategory(
+    value?: string | null,
+  ): 'ANDROID' | 'IOS' | 'OTHER' {
+    if (!value) {
+      return OS_OTHER;
+    }
+
+    const normalized = value.toLowerCase();
+    if (normalized.includes('android')) {
+      return OS_ANDROID;
+    }
+    if (normalized.includes('ios')) {
+      return OS_IOS;
+    }
+
+    return OS_OTHER;
+  }
+}

apps/box-mgnt-api/src/mgnt-backend/feature/ip-report/types/ip-report.types.ts

@@ -0,0 +1,18 @@
+export interface IpLevelHourlyReportRow {
+  /** Epoch second marking the start of the hour bucket. */
+  hourSec: number;
+  /** Channel identifier (optional grouping key). */
+  channelId: string | null;
+  /** Ad type label used for aggregation (AdClickEvents.adType). */
+  adType: string | null;
+  /** Ad object id (AdClickEvents.adsId). */
+  adsId: string | null;
+  /** Login IP (maps to UserLoginHistory.ip). */
+  loginIp: string | null;
+  /** Click IP (maps to AdClickEvents.ip). */
+  clickIp: string | null;
+  /** Normalized OS label derived from UserLoginHistory.os. */
+  osCategory: string | null;
+  /** Aggregated click count; serialize as string for transport, convert to BigInt if needed. */
+  clickCount: string;
+}

apps/box-mgnt-api/src/mgnt-backend/feature/ip-report/validators/ip-report.validators.ts

@@ -0,0 +1,71 @@
+import { Injectable } from '@nestjs/common';
+import { ConfigService } from '@nestjs/config';
+import {
+  ValidationArguments,
+  ValidatorConstraint,
+  ValidatorConstraintInterface,
+} from 'class-validator';
+
+const SECONDS_PER_HOUR = 3600;
+const SECONDS_PER_DAY = 86400;
+
+@ValidatorConstraint({ name: 'IpReportHourAligned', async: false })
+export class IpReportHourAlignedValidator
+  implements ValidatorConstraintInterface
+{
+  validate(value: unknown): boolean {
+    if (typeof value !== 'number' || !Number.isFinite(value)) {
+      return false;
+    }
+
+    return (
+      Number.isInteger(value) && value >= 0 && value % SECONDS_PER_HOUR === 0
+    );
+  }
+
+  defaultMessage(): string {
+    return 'epoch seconds must be an integer aligned to the hour (divisible by 3600)';
+  }
+}
+
+@ValidatorConstraint({ name: 'IpReportWindow', async: false })
+@Injectable()
+export class IpReportWindowValidator implements ValidatorConstraintInterface {
+  constructor(private readonly configService: ConfigService) {}
+
+  validate(toSec: unknown, args: ValidationArguments): boolean {
+    if (typeof toSec !== 'number' || !Number.isFinite(toSec)) {
+      return true;
+    }
+
+    const { fromSec } = args.object as { fromSec?: number };
+    if (typeof fromSec !== 'number' || !Number.isFinite(fromSec)) {
+      return true;
+    }
+
+    const maxWindowHours =
+      this.configService.get<number>('ipReport.maxWindowHours') ?? 48;
+    const rawRetentionDays =
+      this.configService.get<number>('ipReport.rawRetentionDays') ?? 30;
+
+    const windowSeconds = Math.max(1, maxWindowHours) * SECONDS_PER_HOUR;
+    const retentionCutoff =
+      Math.floor(Date.now() / 1000) -
+      Math.max(0, rawRetentionDays) * SECONDS_PER_DAY;
+
+    return (
+      toSec > fromSec &&
+      toSec - fromSec <= windowSeconds &&
+      fromSec >= retentionCutoff
+    );
+  }
+
+  defaultMessage(): string {
+    const maxWindowHours =
+      this.configService.get<number>('ipReport.maxWindowHours') ?? 48;
+    const rawRetentionDays =
+      this.configService.get<number>('ipReport.rawRetentionDays') ?? 30;
+
+    return `time window must be positive, no longer than ${maxWindowHours}h, and no older than ${rawRetentionDays} days`;
+  }
+}

apps/box-mgnt-api/src/mgnt-backend/mgnt-backend.module.ts

@@ -21,6 +21,7 @@ import { HealthModule } from './feature/health/health.module';
 import { CacheSyncModule } from '../cache-sync/cache-sync.module';
 import { ProviderVideoSyncModule } from './feature/provider-video-sync/provider-video-sync.module';
 import { RedisInspectorModule } from './feature/redis-inspector/redis-inspector.module';
+import { IpReportModule } from './feature/ip-report/ip-report.module';
 
 @Module({
   imports: [
@@ -48,6 +49,7 @@ import { RedisInspectorModule } from './feature/redis-inspector/redis-inspector.
           CacheSyncModule,
           RedisInspectorModule,
           ProviderVideoSyncModule,
+          IpReportModule,
         ],
       },
     ]),

apps/box-stats-api/src/feature/stats-events/stats-aggregation.scheduler.ts

@@ -9,6 +9,21 @@ type AggregationResult = {
   errorCount: number;
 };
 
+type RerunRangeCapable = {
+  rerunRange: (args: {
+    fromSec: number;
+    toSec: number;
+    dryRun?: boolean;
+  }) => Promise<unknown>;
+};
+
+type DailyRefreshCapable = {
+  refreshDailyDerivedFromHourly: (args: {
+    fromSec: number;
+    toSec: number;
+  }) => Promise<void>;
+};
+
 @Injectable()
 export class StatsAggregationScheduler implements OnModuleInit {
   private readonly logger = new Logger(StatsAggregationScheduler.name);
@@ -17,7 +32,11 @@ export class StatsAggregationScheduler implements OnModuleInit {
   private windowDays?: number;
 
   // guardrails: avoid overlapping runs + spam
-  private runningAds = false;
+  private runningHourly = false;
+  private runningDaily = false;
+
+  // run a little after boundary so late events can land
+  private runDelaySec = 90;
 
   constructor(
     private readonly configService: ConfigService,
@@ -25,13 +44,11 @@ export class StatsAggregationScheduler implements OnModuleInit {
   ) {}
 
   onModuleInit(): void {
-    // Evaluate config on module init (safer than constructor for startup ordering)
     const enabledRaw = this.configService
       .get<string>('STATS_AGGREGATION_ENABLED')
       ?.trim()
       .toLowerCase();
 
-    // default: enabled (unless explicitly "false" or "0" etc.)
     this.enabled = !['false', '0', 'off', 'no'].includes(enabledRaw ?? '');
 
     const daysRaw = this.configService
@@ -50,11 +67,26 @@ export class StatsAggregationScheduler implements OnModuleInit {
       }
     }
 
+    const delayRaw = this.configService
+      .get<string>('STATS_AGGREGATION_RUN_DELAY_SEC')
+      ?.trim();
+
+    if (delayRaw) {
+      const parsed = Number.parseInt(delayRaw, 10);
+      if (Number.isFinite(parsed) && parsed >= 0 && parsed <= 900) {
+        this.runDelaySec = parsed;
+      } else {
+        this.logger.warn(
+          `Invalid STATS_AGGREGATION_RUN_DELAY_SEC="${delayRaw}" (expected 0..900). Using default ${this.runDelaySec}s.`,
+        );
+      }
+    }
+
     if (this.enabled) {
       this.logger.log(
         `📊 Stats aggregation scheduler enabled (windowDays=${
           this.windowDays ?? 'all time'
-        }, interval=${CronExpression.EVERY_5_MINUTES})`,
+        }, hourly=${CronExpression.EVERY_HOUR}, daily=00:10, delaySec=${this.runDelaySec})`,
       );
     } else {
       this.logger.warn(
@@ -63,43 +95,158 @@ export class StatsAggregationScheduler implements OnModuleInit {
     }
   }
 
-  @Cron(CronExpression.EVERY_5_MINUTES, { name: 'stats-aggregation-ads' })
-  async runAdsAggregation(): Promise<void> {
+  /**
+   * Hourly is the unit of truth.
+   * Runs slightly after the hour, but still processes the previous full hour window.
+   */
+  @Cron(CronExpression.EVERY_HOUR, { name: 'stats-aggregation-hourly' })
+  async runHourly(): Promise<void> {
     if (!this.enabled) return;
 
-    if (this.runningAds) {
+    if (this.runningHourly) {
       this.logger.warn(
-        '⏭️  Skip ads aggregation: previous run still in progress',
+        '⏭️  Skip hourly aggregation: previous run still in progress',
       );
       return;
     }
 
-    this.runningAds = true;
-    const start = Date.now();
+    this.runningHourly = true;
+    const startedAt = Date.now();
+
+    const nowSec = Math.floor(Date.now() / 1000);
+    const effectiveNowSec = nowSec - this.runDelaySec;
 
-    this.logger.log(
-      `⏰ Ads aggregation start (windowDays=${this.windowDays ?? 'all time'})`,
-    );
+    const { fromSec, toSec } = this.prevHourWindow(effectiveNowSec);
+    const tag = `⏰ Hourly aggregation [${fromSec},${toSec})`;
+
+    this.logger.log(`${tag} start`);
 
     try {
-      const result = (await this.statsAggregation.aggregateAdsStats({
-        windowDays: this.windowDays,
-      })) as AggregationResult;
+      // Preferred path: exact hour rerun (idempotent)
+      if (this.hasRerunRange(this.statsAggregation)) {
+        await this.statsAggregation.rerunRange({
+          fromSec,
+          toSec,
+          dryRun: false,
+        });
+        this.logger.log(`${tag} rerunRange done`);
+      } else {
+        // Backward-compatible fallback
+        const result = (await this.statsAggregation.aggregateAdsStats({
+          windowDays: this.windowDays,
+        })) as AggregationResult;
 
-      const ms = Date.now() - start;
-      this.logger.log(
-        `✅ Ads aggregation done in ${ms}ms (${result.successCount}/${result.totalProcessed} updated, ${result.errorCount} errors)`,
+        this.logger.warn(
+          `${tag} used fallback aggregateAdsStats(windowDays=${
+            this.windowDays ?? 'all time'
+          }) (${result.successCount}/${result.totalProcessed} updated, ${result.errorCount} errors)`,
+        );
+      }
+
+      // Optional daily refresh triggered after hourly
+      if (this.hasDailyRefresh(this.statsAggregation)) {
+        await this.statsAggregation.refreshDailyDerivedFromHourly({
+          fromSec,
+          toSec,
+        });
+        this.logger.log(`${tag} daily refresh done`);
+      }
+
+      const ms = Date.now() - startedAt;
+      this.logger.log(`${tag} ✅ done in ${ms}ms`);
+    } catch (err: any) {
+      const ms = Date.now() - startedAt;
+      this.logger.error(
+        `${tag} ❌ failed after ${ms}ms: ${err?.message || String(err)}`,
+        err?.stack,
+      );
+    } finally {
+      this.runningHourly = false;
+    }
+  }
+
+  /**
+   * Daily refresh is derived; safe to run once a day as a "catch-all".
+   * 00:10 (GMT+8 business) — note: Cron uses server TZ; adjust if needed.
+   */
+  @Cron('10 0 * * *', { name: 'stats-aggregation-daily' })
+  async runDailyCatchAll(): Promise<void> {
+    if (!this.enabled) return;
+
+    if (this.runningDaily) {
+      this.logger.warn(
+        '⏭️  Skip daily catch-all: previous run still in progress',
+      );
+      return;
+    }
+
+    if (!this.hasDailyRefresh(this.statsAggregation)) {
+      // No-op if your service doesn’t support daily refresh yet
+      this.logger.warn(
+        'ℹ️ Daily catch-all skipped: refreshDailyDerivedFromHourly not implemented',
       );
-    } catch (err) {
-      const ms = Date.now() - start;
+      return;
+    }
+
+    this.runningDaily = true;
+    const startedAt = Date.now();
+
+    // Refresh “yesterday” (GMT+8 aligned) to catch late events
+    const nowSec = Math.floor(Date.now() / 1000);
+    const yesterdayStartSec = this.floorToDayGmt8(nowSec) - 86400;
+    const yesterdayEndSec = yesterdayStartSec + 86400;
+    const tag = `🗓️ Daily catch-all (GMT+8) [${yesterdayStartSec},${yesterdayEndSec})`;
+
+    this.logger.log(`${tag} start`);
+
+    try {
+      await this.statsAggregation.refreshDailyDerivedFromHourly({
+        fromSec: yesterdayStartSec,
+        toSec: yesterdayEndSec,
+      });
+
+      const ms = Date.now() - startedAt;
+      this.logger.log(`${tag} ✅ done in ${ms}ms`);
+    } catch (err: any) {
+      const ms = Date.now() - startedAt;
       this.logger.error(
-        `❌ Ads aggregation failed after ${ms}ms: ${
-          err instanceof Error ? err.message : String(err)
-        }`,
-        err instanceof Error ? err.stack : undefined,
+        `${tag} ❌ failed after ${ms}ms: ${err?.message || String(err)}`,
+        err?.stack,
       );
     } finally {
-      this.runningAds = false;
+      this.runningDaily = false;
     }
   }
+
+  private prevHourWindow(nowSec: number): { fromSec: number; toSec: number } {
+    const endSec = this.floorToHour(nowSec);
+    return { fromSec: endSec - 3600, toSec: endSec };
+  }
+
+  private floorToHour(sec: number): number {
+    return sec - (sec % 3600);
+  }
+
+  /**
+   * GMT+8 day bucket start in UTC seconds:
+   * shift +8h, floor to day, shift back
+   */
+  private floorToDayGmt8(secUtc: number): number {
+    const shift = 8 * 3600;
+    const shifted = secUtc + shift;
+    const dayStartShifted = shifted - (shifted % 86400);
+    return dayStartShifted - shift;
+  }
+
+  private hasRerunRange(
+    svc: StatsAggregationService,
+  ): svc is StatsAggregationService & RerunRangeCapable {
+    return typeof (svc as any)?.rerunRange === 'function';
+  }
+
+  private hasDailyRefresh(
+    svc: StatsAggregationService,
+  ): svc is StatsAggregationService & DailyRefreshCapable {
+    return typeof (svc as any)?.refreshDailyDerivedFromHourly === 'function';
+  }
 }