feat(analytics): add nightly retention cohort rollup job

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,

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<RetentionCohort>) => 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>(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,
+      );
+    });
+  });
+});

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<AnalyticsEvent>,
+    @InjectRepository(RetentionCohort)
+    private readonly retentionCohortRepository: Repository<RetentionCohort>,
+  ) {}
+
+  /**
+   * 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<void> {
+    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<string[]> {
+    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<CohortUserIdRow>();
+
+    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<number> {
+    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<void> {
+    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;
+}