Merge pull request #596 from bright5455/feature/dau-rollup-jobs

implemented nightly DAU rollup job

Author: A6dulmalik

backend/src/analytics/analytics.module.ts

@@ -2,6 +2,7 @@ import { Module } from '@nestjs/common';
 import { TypeOrmModule } from '@nestjs/typeorm';
 import { AnalyticsEvent } from './entities/analytics-event.entity';
 import { RetentionCohort } from './entities/retention-cohort.entity';
+import { DailyActiveUser } from './entities/daily-active-user.entity';
 import { UsersAnalyticsListener } from './listeners/users-analytics.listener';
 import { AnalyticsController } from './controllers/analytics.controller';
 import { AnalyticsService } from './analytics.service';
@@ -12,7 +13,13 @@ import { GetChurnRiskProvider } from './providers/get-churn-risk.provider';
 import { ExportCsvProvider } from './providers/export-csv.provider';
 
 @Module({
-  imports: [TypeOrmModule.forFeature([AnalyticsEvent, RetentionCohort])],
+  imports: [
+    TypeOrmModule.forFeature([
+      AnalyticsEvent,
+      RetentionCohort,
+      DailyActiveUser,
+    ]),
+  ],
   controllers: [AnalyticsController],
   providers: [
     AnalyticsService,

backend/src/analytics/entities/daily-active-user.entity.ts

@@ -0,0 +1,32 @@
+import {
+  Column,
+  CreateDateColumn,
+  Entity,
+  Index,
+  PrimaryGeneratedColumn,
+  Unique,
+} from 'typeorm';
+
+/**
+ * One row per user who was active on a given day.
+ *
+ * Populated by the nightly `DailyActiveUsersRollupJob` from raw
+ * `AnalyticsEvent` rows, so `GetDauProvider` can compute DAU with a cheap
+ * `COUNT` instead of a `SELECT DISTINCT` scan over raw events.
+ */
+@Entity('daily_active_users')
+@Unique(['date', 'userId'])
+export class DailyActiveUser {
+  @PrimaryGeneratedColumn('uuid')
+  id: string;
+
+  @Index()
+  @Column({ type: 'date' })
+  date: string;
+
+  @Column({ type: 'varchar', length: 100 })
+  userId: string;
+
+  @CreateDateColumn({ type: 'timestamptz' })
+  createdAt: Date;
+}

backend/src/analytics/entities/retention-cohort.entity.ts

@@ -16,7 +16,6 @@ import {
  * on-the-fly from raw `AnalyticsEvent` rows.
  */
 @Entity('retention_cohorts')
-@Index(['cohortDate'])
 export class RetentionCohort {
   @PrimaryGeneratedColumn('uuid')
   id: string;

backend/src/analytics/jobs/daily-active-users-rollup.job.spec.ts

@@ -0,0 +1,131 @@
+import { Logger } from '@nestjs/common';
+import { Test, TestingModule } from '@nestjs/testing';
+import { getRepositoryToken } from '@nestjs/typeorm';
+import { AnalyticsEvent } from '../entities/analytics-event.entity';
+import { DailyActiveUser } from '../entities/daily-active-user.entity';
+import { DailyActiveUsersRollupJob } from './daily-active-users-rollup.job';
+
+describe('DailyActiveUsersRollupJob', () => {
+  let job: DailyActiveUsersRollupJob;
+  const mockAnalyticsEventRepo = {
+    find: jest.fn<Promise<Pick<AnalyticsEvent, 'userId'>[]>, [unknown]>(),
+  };
+  const mockDailyActiveUserRepo = {
+    delete: jest.fn<Promise<unknown>, [unknown]>(),
+    create: jest.fn(
+      (data: Partial<DailyActiveUser>) => data as DailyActiveUser,
+    ),
+    save: jest.fn<Promise<DailyActiveUser[]>, [DailyActiveUser[]]>(),
+  };
+
+  beforeEach(async () => {
+    jest.clearAllMocks();
+
+    const module: TestingModule = await Test.createTestingModule({
+      providers: [
+        DailyActiveUsersRollupJob,
+        {
+          provide: getRepositoryToken(AnalyticsEvent),
+          useValue: mockAnalyticsEventRepo,
+        },
+        {
+          provide: getRepositoryToken(DailyActiveUser),
+          useValue: mockDailyActiveUserRepo,
+        },
+      ],
+    }).compile();
+
+    job = module.get<DailyActiveUsersRollupJob>(DailyActiveUsersRollupJob);
+  });
+
+  it('should be defined', () => {
+    expect(job).toBeDefined();
+  });
+
+  describe('rollupForDate', () => {
+    it('dedupes users who fired multiple events on the same day', async () => {
+      mockAnalyticsEventRepo.find.mockResolvedValue([
+        { userId: 'user-1' },
+        { userId: 'user-1' },
+        { userId: 'user-2' },
+        { userId: 'user-1' },
+        { userId: 'user-3' },
+        { userId: 'user-2' },
+      ] as AnalyticsEvent[]);
+
+      const result = await job.rollupForDate(
+        new Date('2026-07-22T12:00:00.000Z'),
+      );
+
+      expect(result).toEqual({ date: '2026-07-22', rowsWritten: 3 });
+      expect(mockDailyActiveUserRepo.save).toHaveBeenCalledTimes(1);
+
+      const savedRows = mockDailyActiveUserRepo.save.mock.calls[0][0];
+      const savedUserIds = savedRows.map((row) => row.userId).sort();
+      expect(savedUserIds).toEqual(['user-1', 'user-2', 'user-3']);
+      savedRows.forEach((row) => expect(row.date).toBe('2026-07-22'));
+    });
+
+    it('queries raw events for the correct UTC day bounds', async () => {
+      mockAnalyticsEventRepo.find.mockResolvedValue([]);
+
+      await job.rollupForDate(new Date('2026-07-22T12:00:00.000Z'));
+
+      expect(mockAnalyticsEventRepo.find).toHaveBeenCalledWith(
+        expect.objectContaining({
+          select: ['userId'],
+          where: expect.objectContaining({ timestamp: expect.anything() }),
+        }),
+      );
+    });
+
+    it('is idempotent: re-running for the same day replaces rather than duplicates rows', async () => {
+      mockAnalyticsEventRepo.find.mockResolvedValue([
+        { userId: 'user-1' },
+        { userId: 'user-2' },
+      ] as AnalyticsEvent[]);
+
+      await job.rollupForDate(new Date('2026-07-22T00:00:00.000Z'));
+      await job.rollupForDate(new Date('2026-07-22T00:00:00.000Z'));
+
+      expect(mockDailyActiveUserRepo.delete).toHaveBeenCalledTimes(2);
+      expect(mockDailyActiveUserRepo.delete).toHaveBeenNthCalledWith(1, {
+        date: '2026-07-22',
+      });
+      expect(mockDailyActiveUserRepo.delete).toHaveBeenNthCalledWith(2, {
+        date: '2026-07-22',
+      });
+      expect(mockDailyActiveUserRepo.save).toHaveBeenCalledTimes(2);
+    });
+
+    it('writes no rows and skips save when no users were active', async () => {
+      mockAnalyticsEventRepo.find.mockResolvedValue([]);
+
+      const result = await job.rollupForDate(
+        new Date('2026-07-22T00:00:00.000Z'),
+      );
+
+      expect(result.rowsWritten).toBe(0);
+      expect(mockDailyActiveUserRepo.delete).toHaveBeenCalledWith({
+        date: '2026-07-22',
+      });
+      expect(mockDailyActiveUserRepo.save).not.toHaveBeenCalled();
+    });
+
+    it('logs a summary of rows written on completion', async () => {
+      mockAnalyticsEventRepo.find.mockResolvedValue([
+        { userId: 'user-1' },
+      ] as AnalyticsEvent[]);
+      const loggerSpy = jest.spyOn(Logger.prototype, 'log');
+
+      await job.rollupForDate(new Date('2026-07-22T00:00:00.000Z'));
+
+      expect(loggerSpy).toHaveBeenCalledWith(
+        expect.stringContaining('2026-07-22'),
+      );
+      expect(loggerSpy).toHaveBeenCalledWith(
+        expect.stringContaining('1 rows written'),
+      );
+    });
+  });
+});

backend/src/analytics/jobs/daily-active-users-rollup.job.ts

@@ -0,0 +1,66 @@
+import { Injectable, Logger } from '@nestjs/common';
+import { Cron, CronExpression } from '@nestjs/schedule';
+import { InjectRepository } from '@nestjs/typeorm';
+import { Between, Repository } from 'typeorm';
+import { AnalyticsEvent } from '../entities/analytics-event.entity';
+import { DailyActiveUser } from '../entities/daily-active-user.entity';
+
+/**
+ * Nightly rollup that materializes `DailyActiveUser` rows from raw
+ * `AnalyticsEvent` data for the previous day, so DAU can be read cheaply
+ * without scanning raw events on every dashboard load.
+ */
+@Injectable()
+export class DailyActiveUsersRollupJob {
+  private readonly logger = new Logger(DailyActiveUsersRollupJob.name);
+
+  constructor(
+    @InjectRepository(AnalyticsEvent)
+    private readonly analyticsEventRepository: Repository<AnalyticsEvent>,
+    @InjectRepository(DailyActiveUser)
+    private readonly dailyActiveUserRepository: Repository<DailyActiveUser>,
+  ) {}
+
+  @Cron(CronExpression.EVERY_DAY_AT_MIDNIGHT)
+  async handleCron(): Promise<void> {
+    const yesterday = new Date();
+    yesterday.setDate(yesterday.getDate() - 1);
+    await this.rollupForDate(yesterday);
+  }
+
+  /**
+   * Recomputes the `DailyActiveUser` rows for `targetDate`. Safe to re-run
+   * for the same day: existing rows for that date are replaced rather than
+   * appended to.
+   */
+  async rollupForDate(
+    targetDate: Date,
+  ): Promise<{ date: string; rowsWritten: number }> {
+    const dateStr = targetDate.toISOString().split('T')[0];
+    const startOfDay = new Date(`${dateStr}T00:00:00.000Z`);
+    const endOfDay = new Date(`${dateStr}T23:59:59.999Z`);
+
+    const events = await this.analyticsEventRepository.find({
+      select: ['userId'],
+      where: { timestamp: Between(startOfDay, endOfDay) },
+    });
+
+    const distinctUserIds = Array.from(
+      new Set(events.map((event) => event.userId)),
+    );
+
+    await this.dailyActiveUserRepository.delete({ date: dateStr });
+    if (distinctUserIds.length > 0) {
+      const rows = distinctUserIds.map((userId) =>
+        this.dailyActiveUserRepository.create({ date: dateStr, userId }),
+      );
+      await this.dailyActiveUserRepository.save(rows);
+    }
+
+    this.logger.log(
+      `DAU rollup for ${dateStr}: ${distinctUserIds.length} rows written`,
+    );
+
+    return { date: dateStr, rowsWritten: distinctUserIds.length };
+  }
+}

backend/src/app.module.ts

@@ -2,6 +2,7 @@ import { Module, NestModule, MiddlewareConsumer, RequestMethod } from '@nestjs/c
 import { TypeOrmModule } from '@nestjs/typeorm';
 import { ConfigModule, ConfigService } from '@nestjs/config';
 import { EventEmitterModule } from '@nestjs/event-emitter';
+import { ScheduleModule } from '@nestjs/schedule';
 import { RedisModule } from './redis/redis.module';
 import { AuthModule } from './auth/auth.module';
 import appConfig from './config/app.config';
@@ -36,6 +37,7 @@ import { AnalyticsModule } from './analytics/analytics.module';
       load: [appConfig, databaseConfig, jwtConfig],
     }),
     EventEmitterModule.forRoot(),
+    ScheduleModule.forRoot(),
     TypeOrmModule.forRootAsync({
       imports: [ConfigModule],
       inject: [ConfigService],