From bcae39b572cf0d22a4d1640c6d967a20c0ba31a8 Mon Sep 17 00:00:00 2001 From: CIFERBANS123 Date: Sat, 25 Jul 2026 18:34:41 +0000 Subject: [PATCH] feat(caching): improve cache warming with limits, batching, metrics, and deferred startup - Add CACHE_WARM_MAX_ENTRIES, BATCH_SIZE, BATCH_DELAY_MS, STARTUP_DELAY_MS to CACHE_WARMING constants (all env-var configurable with safe defaults) - Prioritise warmCoursesList by enrollment count with recency fallback - Prioritise warmUserProfiles by lastLoginAt recency with updatedAt fallback - Process user profiles in configurable batches with inter-batch delay to bound DB and memory pressure - Emit cache_warming_entries_total (Counter) and cache_warming_duration_seconds (Histogram) Prometheus metrics via MetricsCollectionService - Defer startup warming until after the server is accepting traffic by switching OnModuleInit to OnApplicationBootstrap + STARTUP_DELAY_MS pause - Update specs: MetricsCollectionService mock, enrollment-priority/fallback tests, batch-size assertion with fake timers, deferred-startup delay tests --- src/caching/cache-warming.scheduler.spec.ts | 32 ++- src/caching/cache-warming.scheduler.ts | 43 +++- src/caching/cache-warming.service.spec.ts | 216 +++++++++++++----- src/caching/cache-warming.service.ts | 156 +++++++++---- src/caching/caching.constants.ts | 29 +++ .../metrics/metrics-collection.service.ts | 33 +++ 6 files changed, 412 insertions(+), 97 deletions(-) diff --git a/src/caching/cache-warming.scheduler.spec.ts b/src/caching/cache-warming.scheduler.spec.ts index 69810dec..d3cca4cd 100644 --- a/src/caching/cache-warming.scheduler.spec.ts +++ b/src/caching/cache-warming.scheduler.spec.ts @@ -38,11 +38,29 @@ describe('CacheWarmingScheduler', () => { ); }); - it('runs full warm-up on module init', async () => { - await scheduler.onModuleInit(); + // ── Deferred startup warming ──────────────────────────────────────────────── + + it('runs full warm-up after the startup delay via onApplicationBootstrap', async () => { + jest.useFakeTimers(); + const bootstrapPromise = scheduler.onApplicationBootstrap(); + // Advance past STARTUP_DELAY_MS + await jest.runAllTimersAsync(); + await bootstrapPromise; + jest.useRealTimers(); + expect(warming.warmAll).toHaveBeenCalledTimes(1); }); + it('does NOT call warmAll before the startup delay has elapsed', async () => { + jest.useFakeTimers(); + // Start bootstrap but do not advance timers + void scheduler.onApplicationBootstrap(); + expect(warming.warmAll).not.toHaveBeenCalled(); + jest.useRealTimers(); + }); + + // ── Scheduled methods ─────────────────────────────────────────────────────── + it('delegates scheduled search warming to CacheWarmingService', async () => { await scheduler.warmSearchResults(); expect(warming.warmSearchResults).toHaveBeenCalledTimes(1); @@ -53,6 +71,16 @@ describe('CacheWarmingScheduler', () => { expect(warming.warmUserProfiles).toHaveBeenCalledTimes(1); }); + it('delegates scheduled course list warming to CacheWarmingService', async () => { + await scheduler.warmCoursesList(); + expect(warming.warmCoursesList).toHaveBeenCalledTimes(1); + }); + + it('delegates scheduled popular courses warming to CacheWarmingService', async () => { + await scheduler.warmPopularCourses(); + expect(warming.warmPopularCourses).toHaveBeenCalledTimes(1); + }); + it('publishes cache hit-rate metrics on schedule', () => { scheduler.publishCacheMetrics(); expect(caching.publishHitRateMetrics).toHaveBeenCalledWith('application'); diff --git a/src/caching/cache-warming.scheduler.ts b/src/caching/cache-warming.scheduler.ts index 45ac3873..f2897a43 100644 --- a/src/caching/cache-warming.scheduler.ts +++ b/src/caching/cache-warming.scheduler.ts @@ -1,13 +1,30 @@ -import { Injectable, Logger, OnModuleInit } from '@nestjs/common'; +import { Injectable, Logger, OnApplicationBootstrap } from '@nestjs/common'; import { Cron } from '@nestjs/schedule'; import { CacheWarmingService } from './cache-warming.service'; import { CachingService } from './caching.service'; +import { CACHE_WARMING } from './caching.constants'; /** * Schedules background cache warming for high-traffic query patterns. + * + * ### Startup behaviour + * Implements `OnApplicationBootstrap` instead of `OnModuleInit` so the + * initial warm-up only begins *after* the HTTP server is accepting traffic. + * An additional `CACHE_WARMING.STARTUP_DELAY_MS` pause (default 5 s, tunable + * via `CACHE_WARM_STARTUP_DELAY_MS`) further delays the first pass, giving + * the application time to finish any post-boot work before we fire expensive + * DB queries. + * + * ### Scheduled cadence (aligned with cache TTLs) + * | Target | Cron | TTL | + * |------------------|-----------------|---------| + * | Search results | every 2 min | 2 min | + * | User profiles | every 10 min | 10 min | + * | Course listings | every 15 min | 15 min | + * | Popular courses | every 30 min | 30 min | */ @Injectable() -export class CacheWarmingScheduler implements OnModuleInit { +export class CacheWarmingScheduler implements OnApplicationBootstrap { private readonly logger = new Logger(CacheWarmingScheduler.name); constructor( @@ -15,8 +32,20 @@ export class CacheWarmingScheduler implements OnModuleInit { private readonly caching: CachingService, ) {} - async onModuleInit(): Promise { - this.logger.log('Running initial cache warm-up on startup'); + /** + * Deferred startup warm-up. + * + * Fires after the application is fully bootstrapped (HTTP server is live). + * Waits `CACHE_WARMING.STARTUP_DELAY_MS` milliseconds before running the + * full warm-up so that expensive DB queries do not compete with the last + * steps of the boot sequence. + */ + async onApplicationBootstrap(): Promise { + this.logger.log( + `Startup cache warm-up deferred by ${CACHE_WARMING.STARTUP_DELAY_MS}ms`, + ); + await this.delay(CACHE_WARMING.STARTUP_DELAY_MS); + this.logger.log('Running deferred startup cache warm-up'); await this.runWarmUp('startup'); } @@ -50,6 +79,8 @@ export class CacheWarmingScheduler implements OnModuleInit { this.caching.publishHitRateMetrics('application'); } + // ── Private helpers ───────────────────────────────────────────────────────── + private async runWarmUp( label: string, task?: () => Promise<{ target: string; keysWarmed: number; durationMs: number }>, @@ -68,4 +99,8 @@ export class CacheWarmingScheduler implements OnModuleInit { ); } } + + private delay(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); + } } diff --git a/src/caching/cache-warming.service.spec.ts b/src/caching/cache-warming.service.spec.ts index 9bff0793..7d5003f5 100644 --- a/src/caching/cache-warming.service.spec.ts +++ b/src/caching/cache-warming.service.spec.ts @@ -5,9 +5,10 @@ import { Enrollment } from '../courses/entities/enrollment.entity'; import { User } from '../users/entities/user.entity'; import { ProfileCompletenessService } from '../profile-completeness/profile-completeness.service'; import { SearchService } from '../search/search.service'; +import { MetricsCollectionService } from '../monitoring/metrics/metrics-collection.service'; import { CachingService } from './caching.service'; import { CacheWarmingService } from './cache-warming.service'; -import { CACHE_TTL } from './caching.constants'; +import { CACHE_TTL, CACHE_WARMING } from './caching.constants'; import { buildCourseListKey, buildPopularCoursesKey, @@ -23,6 +24,18 @@ describe('CacheWarmingService', () => { let userRepo: { find: jest.Mock }; let searchService: { search: jest.Mock }; let profileCompleteness: { getScore: jest.Mock }; + let metrics: { recordCacheWarming: jest.Mock }; + + /** Returns a fully-chained query-builder mock whose terminal call resolves to `rows`. */ + const mockQb = (rows: unknown[]) => ({ + select: jest.fn().mockReturnThis(), + addSelect: jest.fn().mockReturnThis(), + innerJoin: jest.fn().mockReturnThis(), + groupBy: jest.fn().mockReturnThis(), + orderBy: jest.fn().mockReturnThis(), + limit: jest.fn().mockReturnThis(), + getRawMany: jest.fn().mockResolvedValue(rows), + }); beforeEach(async () => { caching = { set: jest.fn().mockResolvedValue(undefined) }; @@ -35,6 +48,7 @@ describe('CacheWarmingService', () => { profileCompleteness = { getScore: jest.fn().mockResolvedValue({ score: 80, percentage: 80 }), }; + metrics = { recordCacheWarming: jest.fn() }; const module: TestingModule = await Test.createTestingModule({ providers: [ @@ -42,6 +56,7 @@ describe('CacheWarmingService', () => { { provide: CachingService, useValue: caching }, { provide: SearchService, useValue: searchService }, { provide: ProfileCompletenessService, useValue: profileCompleteness }, + { provide: MetricsCollectionService, useValue: metrics }, { provide: getRepositoryToken(Course), useValue: courseRepo }, { provide: getRepositoryToken(Enrollment), useValue: enrollmentRepo }, { provide: getRepositoryToken(User), useValue: userRepo }, @@ -51,67 +66,166 @@ describe('CacheWarmingService', () => { service = module.get(CacheWarmingService); }); - it('warms published course listings', async () => { - const courses = [{ id: 'c1', status: CourseStatus.PUBLISHED }]; - courseRepo.find.mockResolvedValue(courses); + // ── warmCoursesList ───────────────────────────────────────────────────────── + + describe('warmCoursesList', () => { + it('prioritises courses by enrollment count when enrollment data exists', async () => { + enrollmentRepo.createQueryBuilder.mockReturnValue( + mockQb([{ courseId: 'c1', cnt: '10' }, { courseId: 'c2', cnt: '5' }]), + ); + const courses = [ + { id: 'c1', status: CourseStatus.PUBLISHED }, + { id: 'c2', status: CourseStatus.PUBLISHED }, + ]; + courseRepo.find.mockResolvedValue(courses); + + const result = await service.warmCoursesList(); + + expect(result.target).toBe('COURSES_LIST'); + expect(result.keysWarmed).toBe(1); + expect(caching.set).toHaveBeenCalledWith( + buildCourseListKey('published'), + expect.any(Array), + CACHE_TTL.COURSE_METADATA, + ); + expect(metrics.recordCacheWarming).toHaveBeenCalledWith( + 'COURSES_LIST', + 1, + expect.any(Number), + ); + }); - const result = await service.warmCoursesList(); + it('falls back to recency ordering when no enrollment data exists', async () => { + enrollmentRepo.createQueryBuilder.mockReturnValue(mockQb([])); + const courses = [{ id: 'c1', status: CourseStatus.PUBLISHED }]; + courseRepo.find.mockResolvedValue(courses); - expect(result.target).toBe('COURSES_LIST'); - expect(caching.set).toHaveBeenCalledWith( - buildCourseListKey('published'), - courses, - CACHE_TTL.COURSE_METADATA, - ); + const result = await service.warmCoursesList(); + + expect(result.target).toBe('COURSES_LIST'); + expect(courseRepo.find).toHaveBeenCalledWith( + expect.objectContaining({ + order: { createdAt: 'DESC' }, + take: CACHE_WARMING.MAX_ENTRIES, + }), + ); + }); }); - it('warms popular courses using enrollment counts', async () => { - const qb = { - select: jest.fn().mockReturnThis(), - addSelect: jest.fn().mockReturnThis(), - innerJoin: jest.fn().mockReturnThis(), - groupBy: jest.fn().mockReturnThis(), - orderBy: jest.fn().mockReturnThis(), - limit: jest.fn().mockReturnThis(), - getRawMany: jest.fn().mockResolvedValue([{ courseId: 'c1', enrollmentCount: '10' }]), - }; - enrollmentRepo.createQueryBuilder.mockReturnValue(qb); - courseRepo.find.mockResolvedValue([{ id: 'c1' }]); + // ── warmPopularCourses ────────────────────────────────────────────────────── + + describe('warmPopularCourses', () => { + it('warms popular courses ranked by enrollment count', async () => { + enrollmentRepo.createQueryBuilder.mockReturnValue( + mockQb([{ courseId: 'c1', enrollmentCount: '10' }]), + ); + courseRepo.find.mockResolvedValue([{ id: 'c1' }]); + + const result = await service.warmPopularCourses(); + + expect(result.target).toBe('POPULAR_COURSES'); + expect(caching.set).toHaveBeenCalledWith( + buildPopularCoursesKey(), + [{ id: 'c1' }], + CACHE_TTL.POPULAR_COURSES, + ); + expect(metrics.recordCacheWarming).toHaveBeenCalledWith( + 'POPULAR_COURSES', + 1, + expect.any(Number), + ); + }); - const result = await service.warmPopularCourses(); + it('falls back to recency when enrollment table is empty', async () => { + enrollmentRepo.createQueryBuilder.mockReturnValue(mockQb([])); + courseRepo.find.mockResolvedValue([{ id: 'c2' }]); - expect(result.target).toBe('POPULAR_COURSES'); - expect(caching.set).toHaveBeenCalledWith( - buildPopularCoursesKey(), - [{ id: 'c1' }], - CACHE_TTL.POPULAR_COURSES, - ); + const result = await service.warmPopularCourses(); + + expect(result.target).toBe('POPULAR_COURSES'); + expect(courseRepo.find).toHaveBeenCalledWith( + expect.objectContaining({ where: { status: CourseStatus.PUBLISHED } }), + ); + }); }); - it('warms search result keys for configured queries', async () => { - const result = await service.warmSearchResults(); - - expect(result.target).toBe('SEARCH_RESULTS'); - expect(result.keysWarmed).toBeGreaterThan(0); - expect(searchService.search).toHaveBeenCalled(); - expect(caching.set).toHaveBeenCalledWith( - buildSearchCacheKey(''), - expect.any(Object), - CACHE_TTL.SEARCH_RESULTS, - ); + // ── warmSearchResults ─────────────────────────────────────────────────────── + + describe('warmSearchResults', () => { + it('warms all configured search queries and emits metrics', async () => { + const result = await service.warmSearchResults(); + + expect(result.target).toBe('SEARCH_RESULTS'); + expect(result.keysWarmed).toBeGreaterThan(0); + expect(searchService.search).toHaveBeenCalled(); + expect(caching.set).toHaveBeenCalledWith( + buildSearchCacheKey(''), + expect.any(Object), + CACHE_TTL.SEARCH_RESULTS, + ); + expect(metrics.recordCacheWarming).toHaveBeenCalledWith( + 'SEARCH_RESULTS', + result.keysWarmed, + expect.any(Number), + ); + }); }); - it('warms user profile scores for recently active users', async () => { - userRepo.find.mockResolvedValue([{ id: 'u1' }]); + // ── warmUserProfiles ──────────────────────────────────────────────────────── + + describe('warmUserProfiles', () => { + it('warms recently active users and emits metrics', async () => { + userRepo.find.mockResolvedValue([{ id: 'u1' }]); + + const result = await service.warmUserProfiles(); + + expect(result.target).toBe('USER_PROFILE'); + expect(result.keysWarmed).toBe(1); + expect(profileCompleteness.getScore).toHaveBeenCalledWith('u1'); + expect(caching.set).toHaveBeenCalledWith( + buildUserProfileKey('u1'), + expect.objectContaining({ score: 80 }), + CACHE_TTL.USER_PROFILE, + ); + expect(metrics.recordCacheWarming).toHaveBeenCalledWith( + 'USER_PROFILE', + 1, + expect.any(Number), + ); + }); + + it('falls back to updatedAt ordering when no user has a lastLoginAt', async () => { + // First call (lastLoginAt query) returns empty; second call returns user + userRepo.find + .mockResolvedValueOnce([]) + .mockResolvedValueOnce([{ id: 'u2' }]); + + const result = await service.warmUserProfiles(); + + expect(result.keysWarmed).toBe(1); + // Second find call should order by updatedAt + expect(userRepo.find).toHaveBeenNthCalledWith( + 2, + expect.objectContaining({ order: { updatedAt: 'DESC' } }), + ); + }); + + it('processes users in batches respecting BATCH_SIZE', async () => { + // Produce exactly BATCH_SIZE + 1 users so we exercise the inter-batch delay path + const batchSize = CACHE_WARMING.BATCH_SIZE; + const users = Array.from({ length: batchSize + 1 }, (_, i) => ({ id: `u${i}` })); + userRepo.find.mockResolvedValue(users); - const result = await service.warmUserProfiles(); + // Spy on the private delay via jest fake timers + jest.useFakeTimers(); + const resultPromise = service.warmUserProfiles(); + // Advance past inter-batch delay + await jest.runAllTimersAsync(); + const result = await resultPromise; + jest.useRealTimers(); - expect(result.target).toBe('USER_PROFILE'); - expect(profileCompleteness.getScore).toHaveBeenCalledWith('u1'); - expect(caching.set).toHaveBeenCalledWith( - buildUserProfileKey('u1'), - expect.objectContaining({ score: 80 }), - CACHE_TTL.USER_PROFILE, - ); + expect(result.keysWarmed).toBe(users.length); + expect(profileCompleteness.getScore).toHaveBeenCalledTimes(users.length); + }); }); }); diff --git a/src/caching/cache-warming.service.ts b/src/caching/cache-warming.service.ts index 8f8fcf04..6d2e8511 100644 --- a/src/caching/cache-warming.service.ts +++ b/src/caching/cache-warming.service.ts @@ -6,6 +6,7 @@ import { Enrollment } from '../courses/entities/enrollment.entity'; import { User } from '../users/entities/user.entity'; import { ProfileCompletenessService } from '../profile-completeness/profile-completeness.service'; import { SearchService } from '../search/search.service'; +import { MetricsCollectionService } from '../monitoring/metrics/metrics-collection.service'; import { CachingService } from './caching.service'; import { CACHE_TTL, CACHE_WARMING } from './caching.constants'; import { @@ -23,6 +24,19 @@ export interface WarmResult { /** * Preloads high-traffic cache entries before they are requested. + * + * Key design decisions + * ──────────────────── + * • Every DB read is capped by `CACHE_WARMING.MAX_ENTRIES` so an unbounded + * table never ends up fully in memory. + * • Entries are selected by a *real signal* — enrollment count for courses, + * `lastLoginAt` recency for users — so the warmed set is the set most + * likely to be requested next. + * • Profile warming processes users in configurable batches separated by a + * small async delay, bounding peak Redis write throughput and heap + * allocation per cycle. + * • Each completed warming run is recorded to Prometheus via + * `MetricsCollectionService.recordCacheWarming`. */ @Injectable() export class CacheWarmingService { @@ -32,6 +46,7 @@ export class CacheWarmingService { private readonly caching: CachingService, private readonly searchService: SearchService, private readonly profileCompleteness: ProfileCompletenessService, + private readonly metrics: MetricsCollectionService, @InjectRepository(Course) private readonly courseRepo: Repository, @InjectRepository(Enrollment) @@ -49,25 +64,55 @@ export class CacheWarmingService { ]); } + // ── Courses list ──────────────────────────────────────────────────────────── + async warmCoursesList(): Promise { const started = Date.now(); const key = buildCourseListKey('published'); - const courses = await this.courseRepo.find({ - where: { status: CourseStatus.PUBLISHED }, - order: { createdAt: 'DESC' }, - }); + + // Prioritise by enrollment count so the most popular courses fill the + // list; fall back to recency when the enrollment table is empty. + const topCourseIds = await this.enrollmentRepo + .createQueryBuilder('enrollment') + .select('enrollment.courseId', 'courseId') + .addSelect('COUNT(enrollment.id)', 'cnt') + .innerJoin('enrollment.course', 'course', 'course.status = :status', { + status: CourseStatus.PUBLISHED, + }) + .groupBy('enrollment.courseId') + .orderBy('COUNT(enrollment.id)', 'DESC') + .limit(CACHE_WARMING.MAX_ENTRIES) + .getRawMany<{ courseId: string; cnt: string }>(); + + let courses: Course[]; + if (topCourseIds.length > 0) { + const ids = topCourseIds.map((r) => r.courseId); + const fetched = await this.courseRepo.find({ + where: { id: In(ids), status: CourseStatus.PUBLISHED }, + }); + const rank = new Map(ids.map((id, i) => [id, i])); + courses = fetched.sort((a, b) => (rank.get(a.id) ?? 0) - (rank.get(b.id) ?? 0)); + } else { + courses = await this.courseRepo.find({ + where: { status: CourseStatus.PUBLISHED }, + order: { createdAt: 'DESC' }, + take: CACHE_WARMING.MAX_ENTRIES, + }); + } + await this.caching.set(key, courses, CACHE_TTL.COURSE_METADATA); - this.logger.log(`Warmed courses list (${courses.length} courses)`); - return { - target: 'COURSES_LIST', - keysWarmed: 1, - durationMs: Date.now() - started, - }; + const durationMs = Date.now() - started; + this.logger.log(`Warmed courses list (${courses.length} courses, ${durationMs}ms)`); + this.metrics.recordCacheWarming('COURSES_LIST', 1, durationMs); + return { target: 'COURSES_LIST', keysWarmed: 1, durationMs }; } + // ── Popular courses ───────────────────────────────────────────────────────── + async warmPopularCourses(): Promise { const started = Date.now(); const key = buildPopularCoursesKey(); + const popular = await this.enrollmentRepo .createQueryBuilder('enrollment') .select('enrollment.courseId', 'courseId') @@ -77,7 +122,8 @@ export class CacheWarmingService { }) .groupBy('enrollment.courseId') .orderBy('COUNT(enrollment.id)', 'DESC') - .limit(CACHE_WARMING.POPULAR_COURSES_LIMIT) + // Respect the configurable ceiling but never exceed the dedicated limit + .limit(Math.min(CACHE_WARMING.POPULAR_COURSES_LIMIT, CACHE_WARMING.MAX_ENTRIES)) .getRawMany<{ courseId: string; enrollmentCount: string }>(); let courses: Course[]; @@ -85,7 +131,7 @@ export class CacheWarmingService { courses = await this.courseRepo.find({ where: { status: CourseStatus.PUBLISHED }, order: { createdAt: 'DESC' }, - take: CACHE_WARMING.POPULAR_COURSES_LIMIT, + take: Math.min(CACHE_WARMING.POPULAR_COURSES_LIMIT, CACHE_WARMING.MAX_ENTRIES), }); } else { const courseIds = popular.map((row) => row.courseId); @@ -95,14 +141,14 @@ export class CacheWarmingService { } await this.caching.set(key, courses, CACHE_TTL.POPULAR_COURSES); - this.logger.log(`Warmed popular courses (${courses.length} courses)`); - return { - target: 'POPULAR_COURSES', - keysWarmed: 1, - durationMs: Date.now() - started, - }; + const durationMs = Date.now() - started; + this.logger.log(`Warmed popular courses (${courses.length} courses, ${durationMs}ms)`); + this.metrics.recordCacheWarming('POPULAR_COURSES', 1, durationMs); + return { target: 'POPULAR_COURSES', keysWarmed: 1, durationMs }; } + // ── Search results ────────────────────────────────────────────────────────── + async warmSearchResults(): Promise { const started = Date.now(); let keysWarmed = 0; @@ -119,42 +165,72 @@ export class CacheWarmingService { await this.caching.set(filtersKey, filteredResult, CACHE_TTL.SEARCH_RESULTS); keysWarmed += 1; - this.logger.log(`Warmed search results (${keysWarmed} keys)`); - return { - target: 'SEARCH_RESULTS', - keysWarmed, - durationMs: Date.now() - started, - }; + const durationMs = Date.now() - started; + this.logger.log(`Warmed search results (${keysWarmed} keys, ${durationMs}ms)`); + this.metrics.recordCacheWarming('SEARCH_RESULTS', keysWarmed, durationMs); + return { target: 'SEARCH_RESULTS', keysWarmed, durationMs }; } + // ── User profiles ─────────────────────────────────────────────────────────── + + /** + * Warms profile-completeness scores for the most recently active users. + * + * Users are ranked by `lastLoginAt DESC` (real activity signal) and capped + * by `CACHE_WARMING.MAX_ENTRIES`. The working set is then processed in + * fixed-size batches (`CACHE_WARMING.BATCH_SIZE`), with a short async delay + * (`CACHE_WARMING.BATCH_DELAY_MS`) between batches. This keeps peak Redis + * write throughput and heap allocation bounded even when MAX_ENTRIES is + * set very high. + */ async warmUserProfiles(): Promise { const started = Date.now(); + + // Prefer recently active users; fall back to recently updated rows when + // no one has a non-null lastLoginAt (e.g. fresh installs, seed data). let users = await this.userRepo.find({ where: { lastLoginAt: Not(IsNull()) }, order: { lastLoginAt: 'DESC' }, - take: CACHE_WARMING.USER_PROFILE_WARM_LIMIT, + take: CACHE_WARMING.MAX_ENTRIES, }); if (users.length === 0) { users = await this.userRepo.find({ order: { updatedAt: 'DESC' }, - take: CACHE_WARMING.USER_PROFILE_WARM_LIMIT, + take: CACHE_WARMING.MAX_ENTRIES, }); } - await Promise.all( - users.map(async (user) => { - const profile = await this.profileCompleteness.getScore(user.id); - const key = buildUserProfileKey(user.id); - await this.caching.set(key, profile, CACHE_TTL.USER_PROFILE); - }), - ); - - this.logger.log(`Warmed user profiles (${users.length} users)`); - return { - target: 'USER_PROFILE', - keysWarmed: users.length, - durationMs: Date.now() - started, - }; + let keysWarmed = 0; + + for (let i = 0; i < users.length; i += CACHE_WARMING.BATCH_SIZE) { + const batch = users.slice(i, i + CACHE_WARMING.BATCH_SIZE); + + await Promise.all( + batch.map(async (user) => { + const profile = await this.profileCompleteness.getScore(user.id); + const key = buildUserProfileKey(user.id); + await this.caching.set(key, profile, CACHE_TTL.USER_PROFILE); + }), + ); + + keysWarmed += batch.length; + + // Yield the event-loop between batches; skip the delay after the last one. + if (i + CACHE_WARMING.BATCH_SIZE < users.length) { + await this.delay(CACHE_WARMING.BATCH_DELAY_MS); + } + } + + const durationMs = Date.now() - started; + this.logger.log(`Warmed user profiles (${keysWarmed} users, ${durationMs}ms)`); + this.metrics.recordCacheWarming('USER_PROFILE', keysWarmed, durationMs); + return { target: 'USER_PROFILE', keysWarmed, durationMs }; + } + + // ── Helpers ───────────────────────────────────────────────────────────────── + + private delay(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); } } diff --git a/src/caching/caching.constants.ts b/src/caching/caching.constants.ts index 50b096dd..7fcf2230 100644 --- a/src/caching/caching.constants.ts +++ b/src/caching/caching.constants.ts @@ -59,9 +59,38 @@ export const CACHE_PREFIXES = { QUIZ: 'cache:quiz', } as const; export const CACHE_WARMING = { + /** Hard ceiling on rows fetched per warming target. Override with CACHE_WARM_MAX_ENTRIES. */ POPULAR_COURSES_LIMIT: 20, USER_PROFILE_WARM_LIMIT: 50, SEARCH_WARM_QUERIES: ['', 'programming', 'math', 'science'], + + /** + * Maximum number of cache entries to warm per target. + * Applies as a `take` to every DB read so we never pull an unbounded + * result set into memory. Can be overridden via the CACHE_WARM_MAX_ENTRIES + * environment variable. + */ + MAX_ENTRIES: parseInt(process.env.CACHE_WARM_MAX_ENTRIES ?? '500', 10), + + /** + * Number of entries processed in a single batch before the inter-batch + * delay is applied. Bounds memory pressure per cycle. + */ + BATCH_SIZE: parseInt(process.env.CACHE_WARM_BATCH_SIZE ?? '50', 10), + + /** + * Milliseconds to wait between consecutive batches. + * A small pause releases the event-loop and avoids overwhelming Redis + * with burst writes. + */ + BATCH_DELAY_MS: parseInt(process.env.CACHE_WARM_BATCH_DELAY_MS ?? '50', 10), + + /** + * How long (ms) to wait after the NestJS application has signalled it is + * ready before the first cache warm-up is triggered. Allows the HTTP + * server to start accepting traffic before we run expensive DB queries. + */ + STARTUP_DELAY_MS: parseInt(process.env.CACHE_WARM_STARTUP_DELAY_MS ?? '5000', 10), } as const; export const CACHE_EVENTS = { diff --git a/src/monitoring/metrics/metrics-collection.service.ts b/src/monitoring/metrics/metrics-collection.service.ts index deba2142..8c7f87e8 100644 --- a/src/monitoring/metrics/metrics-collection.service.ts +++ b/src/monitoring/metrics/metrics-collection.service.ts @@ -86,6 +86,12 @@ export class MetricsCollectionService implements OnModuleInit { /** Cache hit rate percentage, labelled by cache_type */ public cacheHitRate: Gauge; + /** Total cache entries warmed, labelled by target (e.g. COURSES_LIST) */ + public cacheWarmingEntriesTotal: Counter; + + /** Duration of a single cache warming run in seconds, labelled by target */ + public cacheWarmingDuration: Histogram; + // ── Business Metrics – Queues ────────────────────────────────────────────── /** Queue job processing duration, labelled by queue_name and job_type */ @@ -224,6 +230,18 @@ export class MetricsCollectionService implements OnModuleInit { this.cacheHitRate.set({ cache_type: cacheType }, hitRate); } + /** + * Records one completed cache warming run. + * + * @param target Warming target label (e.g. 'COURSES_LIST') + * @param keysWarmed Number of cache entries written + * @param durationMs Wall-clock duration of the warming run in milliseconds + */ + recordCacheWarming(target: string, keysWarmed: number, durationMs: number): void { + this.cacheWarmingEntriesTotal.inc({ target }, keysWarmed); + this.cacheWarmingDuration.observe({ target }, durationMs / 1000); + } + // ── Recording helpers – Queues ──────────────────────────────────────────── recordQueueProcessingTime(queueName: string, jobType: string, duration: number): void { @@ -434,6 +452,21 @@ export class MetricsCollectionService implements OnModuleInit { registers: [this.registry], }); + this.cacheWarmingEntriesTotal = new Counter({ + name: 'cache_warming_entries_total', + help: 'Total number of cache entries warmed per warming target', + labelNames: ['target'], + registers: [this.registry], + }); + + this.cacheWarmingDuration = new Histogram({ + name: 'cache_warming_duration_seconds', + help: 'Duration of a single cache warming run in seconds per target', + labelNames: ['target'], + buckets: [0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30], + registers: [this.registry], + }); + // Queues this.queueProcessingTime = new Histogram({ name: 'queue_processing_duration_seconds',