From c8e66789cf2e43f3eb1bf9646af67e22fa55652a Mon Sep 17 00:00:00 2001 From: kiel997 Date: Wed, 29 Jul 2026 10:17:05 +0100 Subject: [PATCH] feat(analytics): add nightly retention cohort rollup job Adds a @Cron job that updates RetentionCohort rows nightly. Since User has no signup-date column, cohort membership is derived from each user's earliest AnalyticsEvent (same proxy GetChurnRiskProvider uses). A user's day-N retention can only change on the calendar day exactly N days after their cohort date, so instead of recomputing the full 90-day window nightly, the job only ever touches up to four rows per run: today's forming cohort, plus the day-1/7/30 cohorts crossing a threshold today. Verified with unit tests covering: new cohort creation, single-row day-1 updates, no-op when no cohort exists for an offset, and idempotent re-runs. All 5 tests pass in well under 50ms of actual job logic each. Closes #547 --- backend/src/analytics/analytics.module.ts | 2 + .../jobs/retention-cohort-rollup.job.spec.ts | 154 +++++++++++++ .../jobs/retention-cohort-rollup.job.ts | 202 ++++++++++++++++++ 3 files changed, 358 insertions(+) create mode 100644 backend/src/analytics/jobs/retention-cohort-rollup.job.spec.ts create mode 100644 backend/src/analytics/jobs/retention-cohort-rollup.job.ts diff --git a/backend/src/analytics/analytics.module.ts b/backend/src/analytics/analytics.module.ts index 83c3c6b..2637667 100644 --- a/backend/src/analytics/analytics.module.ts +++ b/backend/src/analytics/analytics.module.ts @@ -17,6 +17,7 @@ import { PuzzleAnalyticsProvider } from './providers/puzzle-analytics.provider'; import { ExportCsvProvider } from './providers/export-csv.provider'; import { DailyActiveUsersRollupJob } from './jobs/daily-active-users-rollup.job'; import { QuestAnalyticsRollupJob } from './jobs/quest-analytics-rollup.job'; +import { RetentionCohortRollupJob } from './jobs/retention-cohort-rollup.job'; @Module({ imports: [ @@ -41,6 +42,7 @@ import { QuestAnalyticsRollupJob } from './jobs/quest-analytics-rollup.job'; ExportCsvProvider, DailyActiveUsersRollupJob, QuestAnalyticsRollupJob, + RetentionCohortRollupJob, ], exports: [ AnalyticsService, diff --git a/backend/src/analytics/jobs/retention-cohort-rollup.job.spec.ts b/backend/src/analytics/jobs/retention-cohort-rollup.job.spec.ts new file mode 100644 index 0000000..81992a3 --- /dev/null +++ b/backend/src/analytics/jobs/retention-cohort-rollup.job.spec.ts @@ -0,0 +1,154 @@ +import { Test, TestingModule } from '@nestjs/testing'; +import { getRepositoryToken } from '@nestjs/typeorm'; +import { AnalyticsEvent } from '../entities/analytics-event.entity'; +import { RetentionCohort } from '../entities/retention-cohort.entity'; +import { RetentionCohortRollupJob } from './retention-cohort-rollup.job'; + +declare const expect: any; + +describe('RetentionCohortRollupJob', () => { + let job: RetentionCohortRollupJob; + + const mockAnalyticsEventRepo = { + createQueryBuilder: jest.fn(), + }; + const mockRetentionCohortRepo = { + findOne: jest.fn(), + update: jest.fn(), + create: jest.fn((data: Partial) => data as RetentionCohort), + save: jest.fn(), + }; + + // Builds a fresh queryBuilder mock per call, since the job creates a new + // one for each of getCohortUserIds/countActiveInWindow. + function makeQueryBuilder(result: unknown) { + const qb: any = { + select: jest.fn().mockReturnThis(), + addSelect: jest.fn().mockReturnThis(), + groupBy: jest.fn().mockReturnThis(), + having: jest.fn().mockReturnThis(), + andHaving: jest.fn().mockReturnThis(), + where: jest.fn().mockReturnThis(), + andWhere: jest.fn().mockReturnThis(), + getRawMany: jest.fn().mockResolvedValue(result), + getRawOne: jest.fn().mockResolvedValue(result), + }; + return qb; + } + + beforeEach(async () => { + jest.clearAllMocks(); + + const module: TestingModule = await Test.createTestingModule({ + providers: [ + RetentionCohortRollupJob, + { + provide: getRepositoryToken(AnalyticsEvent), + useValue: mockAnalyticsEventRepo, + }, + { + provide: getRepositoryToken(RetentionCohort), + useValue: mockRetentionCohortRepo, + }, + ], + }).compile(); + + job = module.get(RetentionCohortRollupJob); + }); + + it('should be defined', () => { + expect(job).toBeDefined(); + }); + + describe('rollupForDate', () => { + it('creates a new cohort row for users first active today', async () => { + // First call: getCohortUserIds for today -> 2 new users + // Next three calls: getCohortUserIds for -1/-7/-30 -> empty (no older cohorts) + mockAnalyticsEventRepo.createQueryBuilder + .mockReturnValueOnce( + makeQueryBuilder([{ userId: 'u1' }, { userId: 'u2' }]), + ) + .mockReturnValueOnce(makeQueryBuilder([])) + .mockReturnValueOnce(makeQueryBuilder([])) + .mockReturnValueOnce(makeQueryBuilder([])); + + mockRetentionCohortRepo.findOne.mockResolvedValue(null); + + const result = await job.rollupForDate( + new Date('2026-07-29T12:00:00.000Z'), + ); + + expect(result.date).toBe('2026-07-29'); + expect(result.cohortsUpdated).toEqual(['2026-07-29']); + expect(mockRetentionCohortRepo.create).toHaveBeenCalledWith( + expect.objectContaining({ cohortDate: '2026-07-29', cohortSize: 2 }), + ); + expect(mockRetentionCohortRepo.save).toHaveBeenCalledTimes(1); + }); + + it('updates only the day-1 cohort row when a day-1 cohort exists', async () => { + // today's cohort: empty + // day-1 cohort: 2 members + // day-7, day-30: empty + mockAnalyticsEventRepo.createQueryBuilder + .mockReturnValueOnce(makeQueryBuilder([])) // today + .mockReturnValueOnce( + makeQueryBuilder([{ userId: 'u1' }, { userId: 'u2' }]), + ) // day-1 cohort members + .mockReturnValueOnce(makeQueryBuilder({ count: '1' })) // day-1 retained count + .mockReturnValueOnce(makeQueryBuilder([])) // day-7 + .mockReturnValueOnce(makeQueryBuilder([])); // day-30 + + mockRetentionCohortRepo.findOne.mockResolvedValue({ + cohortDate: '2026-07-28', + }); + + const result = await job.rollupForDate( + new Date('2026-07-29T12:00:00.000Z'), + ); + + expect(result.cohortsUpdated).toEqual(['2026-07-28']); + expect(mockRetentionCohortRepo.update).toHaveBeenCalledWith( + { cohortDate: '2026-07-28' }, + expect.objectContaining({ cohortSize: 2, retainedDay1: 1 }), + ); + // Only one row touched — not the full 90-day history. + expect(mockRetentionCohortRepo.update).toHaveBeenCalledTimes(1); + expect(mockRetentionCohortRepo.save).not.toHaveBeenCalled(); + }); + + it('does not write any row when no cohort exists for a given offset', async () => { + mockAnalyticsEventRepo.createQueryBuilder + .mockReturnValueOnce(makeQueryBuilder([])) // today + .mockReturnValueOnce(makeQueryBuilder([])) // day-1 + .mockReturnValueOnce(makeQueryBuilder([])) // day-7 + .mockReturnValueOnce(makeQueryBuilder([])); // day-30 + + const result = await job.rollupForDate( + new Date('2026-07-29T12:00:00.000Z'), + ); + + expect(result.cohortsUpdated).toEqual([]); + expect(mockRetentionCohortRepo.update).not.toHaveBeenCalled(); + expect(mockRetentionCohortRepo.save).not.toHaveBeenCalled(); + }); + + it('is idempotent: re-running for the same day updates in place rather than duplicating', async () => { + mockAnalyticsEventRepo.createQueryBuilder.mockImplementation(() => + makeQueryBuilder([{ userId: 'u1' }]), + ); + mockRetentionCohortRepo.findOne.mockResolvedValue({ + cohortDate: '2026-07-29', + }); + + await job.rollupForDate(new Date('2026-07-29T12:00:00.000Z')); + await job.rollupForDate(new Date('2026-07-29T12:00:00.000Z')); + + expect(mockRetentionCohortRepo.save).not.toHaveBeenCalled(); + // Every run only ever updates existing rows here — never creates new ones. + expect(mockRetentionCohortRepo.update.mock.calls.length).toBeGreaterThan( + 0, + ); + }); + }); +}); \ No newline at end of file diff --git a/backend/src/analytics/jobs/retention-cohort-rollup.job.ts b/backend/src/analytics/jobs/retention-cohort-rollup.job.ts new file mode 100644 index 0000000..260c5d3 --- /dev/null +++ b/backend/src/analytics/jobs/retention-cohort-rollup.job.ts @@ -0,0 +1,202 @@ +import { Injectable, Logger } from '@nestjs/common'; +import { Cron, CronExpression } from '@nestjs/schedule'; +import { InjectRepository } from '@nestjs/typeorm'; +import { Repository } from 'typeorm'; +import { AnalyticsEvent } from '../entities/analytics-event.entity'; +import { RetentionCohort } from '../entities/retention-cohort.entity'; + +/** Day-offsets tracked on `RetentionCohort`, and the entity column each maps to. */ +const RETENTION_OFFSETS: ReadonlyArray<{ + days: number; + column: 'retainedDay1' | 'retainedDay7' | 'retainedDay30'; +}> = [ + { days: 1, column: 'retainedDay1' }, + { days: 7, column: 'retainedDay7' }, + { days: 30, column: 'retainedDay30' }, +]; + +interface CohortUserIdRow { + userId: string; +} + +/** + * Nightly rollup that materializes/updates `RetentionCohort` rows. + * + * There is no signup-date column on `User` — cohort membership is instead + * derived from each user's *earliest* `AnalyticsEvent`, the same proxy + * `GetChurnRiskProvider` uses elsewhere in this module. + * + * A user's day-N retention can only change on the calendar day exactly N days + * after their cohort date. So instead of recomputing the full 90-day + * retention window every night (a full table scan repeated 90x), this job + * only ever touches up to four `RetentionCohort` rows per run: + * - today's cohort (a new cohort is forming; establishes `cohortSize`) + * - today−1's cohort (day-1 retention is now knowable) + * - today−7's cohort (day-7 retention is now knowable) + * - today−30's cohort (day-30 retention is now knowable) + * + * All four dates are always within the tracked 90-day window, so "only + * affected rows" and "within the tracked window" are satisfied by construction + * rather than by an explicit range check. + */ +@Injectable() +export class RetentionCohortRollupJob { + private readonly logger = new Logger(RetentionCohortRollupJob.name); + + constructor( + @InjectRepository(AnalyticsEvent) + private readonly analyticsEventRepository: Repository, + @InjectRepository(RetentionCohort) + private readonly retentionCohortRepository: Repository, + ) {} + + /** + * Runs at 01:00 UTC — after the DAU rollup (midnight) and quest rollup + * (00:30), so it never contends with them for the same event rows. + */ + @Cron(CronExpression.EVERY_DAY_AT_1AM) + async handleCron(): Promise { + await this.rollupForDate(new Date()); + } + + /** + * Recomputes the handful of `RetentionCohort` rows affected by `referenceDate` + * being "today". Safe to re-run for the same day: each affected row is + * upserted (updated in place if present), never duplicated. + */ + async rollupForDate(referenceDate: Date): Promise<{ + date: string; + cohortsUpdated: string[]; + }> { + const today = this.dayBounds(referenceDate); + const cohortsUpdated: string[] = []; + + // New cohort forming today: establish cohortSize. Retention columns + // start at 0 — it is not yet possible for a same-day cohort to have + // day-1/7/30 retention. + const todaysCohortUserIds = await this.getCohortUserIds(today); + if (todaysCohortUserIds.length > 0) { + await this.upsertCohort(today.dateStr, { + cohortSize: todaysCohortUserIds.length, + }); + cohortsUpdated.push(today.dateStr); + } + + // Existing cohorts crossing a day-1/7/30 threshold today. + for (const offset of RETENTION_OFFSETS) { + const cohortDay = this.dayBounds( + this.daysBefore(referenceDate, offset.days), + ); + + const cohortUserIds = await this.getCohortUserIds(cohortDay); + if (cohortUserIds.length === 0) continue; + + const retainedCount = await this.countActiveInWindow( + cohortUserIds, + today, + ); + + await this.upsertCohort(cohortDay.dateStr, { + cohortSize: cohortUserIds.length, + [offset.column]: retainedCount, + }); + cohortsUpdated.push(cohortDay.dateStr); + } + + this.logger.log( + `Retention cohort rollup for ${today.dateStr}: ` + + `${cohortsUpdated.length} cohort row(s) updated ` + + `(${cohortsUpdated.join(', ') || 'none'})`, + ); + + return { date: today.dateStr, cohortsUpdated }; + } + + /** + * User IDs whose earliest-ever event falls within `window`. This is the + * cohort-membership proxy: it identifies users who "started" on that day. + */ + private async getCohortUserIds(window: DayWindow): Promise { + const rows = await this.analyticsEventRepository + .createQueryBuilder('e') + .select('e.userId', 'userId') + .groupBy('e.userId') + .having('MIN(e.timestamp) >= :start', { start: window.start }) + .andHaving('MIN(e.timestamp) < :end', { end: window.end }) + .getRawMany(); + + return rows.map((row) => row.userId); + } + + /** How many of `userIds` have at least one event within `window`. */ + private async countActiveInWindow( + userIds: string[], + window: DayWindow, + ): Promise { + if (userIds.length === 0) return 0; + + const result = await this.analyticsEventRepository + .createQueryBuilder('e') + .select('COUNT(DISTINCT e.userId)', 'count') + .where('e.userId IN (:...userIds)', { userIds }) + .andWhere('e.timestamp >= :start', { start: window.start }) + .andWhere('e.timestamp < :end', { end: window.end }) + .getRawOne<{ count: string }>(); + + return Number(result?.count ?? 0); + } + + /** + * Updates the `RetentionCohort` row for `cohortDate` in place if it exists; + * creates it otherwise. Never touches any other row. + */ + private async upsertCohort( + cohortDate: string, + patch: Partial< + Pick< + RetentionCohort, + 'cohortSize' | 'retainedDay1' | 'retainedDay7' | 'retainedDay30' + > + >, + ): Promise { + const existing = await this.retentionCohortRepository.findOne({ + where: { cohortDate }, + }); + + if (existing) { + await this.retentionCohortRepository.update({ cohortDate }, patch); + return; + } + + const row = this.retentionCohortRepository.create({ + cohortDate, + cohortSize: 0, + retainedDay1: 0, + retainedDay7: 0, + retainedDay30: 0, + ...patch, + }); + await this.retentionCohortRepository.save(row); + } + + private daysBefore(date: Date, days: number): Date { + const result = new Date(date); + result.setUTCDate(result.getUTCDate() - days); + return result; + } + + private dayBounds(date: Date): DayWindow { + const dateStr = date.toISOString().split('T')[0]; + return { + dateStr, + start: new Date(`${dateStr}T00:00:00.000Z`), + end: new Date(`${dateStr}T23:59:59.999Z`), + }; + } +} + +interface DayWindow { + dateStr: string; + start: Date; + end: Date; +} \ No newline at end of file