REFRESH-AHEAD

Introduction

This article will guide you on how to implement REFRESH-AHEAD on NestJS with Redis and Postgres.
We will need to use @nestjs/cache-manager to create a service that handles keys about to expire.
We will also use @nestjs/bullmq for background processing to query the database and update the cache.
For theoretical details, you can review the previous article where I covered this topic.

Detail

Create the file constant/index.ts with Queue, Job and Function as follows:

export enum QueueName {
  CACHE_REFRESH = 'cache-refresh',
}

export enum JobName {
  REFRESH_KEY = 'refresh-key',
}

export enum FunctionName {
  PRODUCT_DETAIL = 'product-detail',
  TRENDING_PRODUCT = 'trending-product',
}

Create file module/bullmq.module.ts

import {BullModule} from '@nestjs/bullmq'
import {Module} from '@nestjs/common'
import {ConfigModule, ConfigService} from '@nestjs/config'
import {QueueName} from 'src/constant'

@Module({
  imports: [
    BullModule.forRootAsync({
      imports: [ConfigModule],
      inject: [ConfigService],
      useFactory: (configService: ConfigService) => ({
        connection: {
          host: configService.get<string>('VALKEY_HOST'),
          port: configService.get<number>('VALKEY_PORT'),
        },
      }),
    }),
    BullModule.registerQueue(
      {
        name: QueueName.CACHE_REFRESH,
      }
    ),
  ],
  exports: [
    BullModule,
  ],
})
export class BullMqModule {}

Create file service/refresh-ahead.service.ts

import {InjectQueue} from '@nestjs/bullmq'
import {CACHE_MANAGER} from '@nestjs/cache-manager'
import {Inject, Injectable} from '@nestjs/common'
import {Queue} from 'bullmq'
import type {Cache} from 'cache-manager'
import {desc, eq} from 'drizzle-orm'
import type {NodePgDatabase} from 'drizzle-orm/node-postgres'
import {FunctionName, JobName, QueueName} from 'src/constant'
import * as schema from 'src/drizzle-orm/schema/schema'

interface CachedPayload<T> {
  data: T
  expireAt: number 
}

@Injectable()
export class RefreshAheadCacheService {
  constructor(
    @Inject('DRIZZLE_DB') private readonly db: NodePgDatabase<typeof schema>,
    @Inject(CACHE_MANAGER) private cacheManager: Cache,
    @InjectQueue(QueueName.CACHE_REFRESH) private refreshQueue: Queue
  ) {}

  async getOrRefresh<T>(
    key: string,
    fetchFnName: string, 
    fetchArgs: any[], 
    ttlMs: number = 60000, 
    refreshThresholdRatio: number = 0.2 
  ): Promise<T> {
    const cached = await this.cacheManager.get<CachedPayload<T>>(key)
    const now = Date.now()

    if (cached) {
      const remainingTime = cached.expireAt - now
      const refreshWindow = ttlMs * refreshThresholdRatio

      if (remainingTime > 0 && remainingTime <= refreshWindow) {
        this.refreshQueue
          .add(
            JobName.REFRESH_KEY,
            {key, fetchFnName, fetchArgs, ttlMs},
            {
              jobId: `refresh:${key}:${Math.floor(now / 10000)}`,
              removeOnComplete: true,
            }
          )
          .catch(() => {}) 
      }

      return cached.data
    }

    return this.executeAndSetCache(key, fetchFnName, fetchArgs, ttlMs)
  }

  public async executeAndSetCache<T>(
    key: string,
    fetchFnName: string,
    fetchArgs: any[],
    ttlMs: number
  ) {
    const data = (await this.resolveDbFetch(fetchFnName, fetchArgs)) as T

    const payload: CachedPayload<T> = {
      data,
      expireAt: Date.now() + ttlMs,
    }

    await this.cacheManager.set(key, payload, ttlMs)
    return data
  }

  private resolveDbFetch(fnName: string, args: any[]) {
    switch (fnName) {
      case FunctionName.PRODUCT_DETAIL:
        return this.db
          .select()
          .from(schema.products)
          .where(eq(schema.products.id, args[0]))
          .then(res => res[0])
      case FunctionName.TRENDING_PRODUCT:
        return this.db
          .select()
          .from(schema.products)
          .orderBy(desc(schema.products.views)) 
          .limit(10) 
    }
  }
}

The getOrRefresh function checks whether the TTL matches the window time to create a Job for querying data from the database again.
The executeAndSetCache function handles creating the cached data.
The resolveDbFetch function works according to the function and arguments to query the appropriate data.

Create file worker/cache-refresh.processor.ts. This is where background jobs are processed, simply calling executeAndSetCache is sufficient.

import {Processor, WorkerHost} from '@nestjs/bullmq'
import {Injectable} from '@nestjs/common'
import {Job} from 'bullmq'
import {QueueName} from 'src/constant'
import {RefreshAheadCacheService} from 'src/service/refresh-ahead.service'

@Processor(QueueName.CACHE_REFRESH)
@Injectable()
export class CacheRefreshProcessor extends WorkerHost {
  constructor(private refreshAheadService: RefreshAheadCacheService) {
    super()
  }

  async process(job: Job<any>): Promise<any> {
    const {key, fetchFnName, fetchArgs, ttlMs} = job.data

    await this.refreshAheadService.executeAndSetCache(
      key,
      fetchFnName,
      fetchArgs,
      ttlMs
    )
  }
}

Create file service/product.service.ts. Just use the getOrRefresh function to pass the function and arguments. You can also customize TTL and refreshThresholdRatio for each individual case.

import {Injectable} from '@nestjs/common'
import {FunctionName} from 'src/constant'
import type {Product} from 'src/dto/product2.dto'
import {RefreshAheadCacheService} from './refresh-ahead.service'

@Injectable()
export class ProductService {
  constructor(private readonly refreshAheadCache: RefreshAheadCacheService) {}

  async getProductById(id: number) {
    return this.refreshAheadCache.getOrRefresh<Product>(
      `product_${id}`,
      FunctionName.PRODUCT_DETAIL, 
      [id] 
    )
  }

  async getTrendingProducts() {
    return this.refreshAheadCache.getOrRefresh<Product[]>(
      'products:trending',
      FunctionName.TRENDING_PRODUCT, 
      [], 
      10 * 60 * 1000, 
      0.3 
    )
  }
}

Create file controller/product.controller.ts

import {Controller, Get, Param, ParseIntPipe} from '@nestjs/common'
import {ProductService} from 'src/service/product.service'

@Controller('products')
export class ProductsController {
  constructor(private readonly productService: ProductService) {}

  @Get('trending')
  async getTrendingProducts() {
    return await this.productService.getTrendingProducts()
  }

  @Get(':id')
  async getProduct(@Param('id', ParseIntPipe) id: number) {
    return await this.productService.getProductById(id)
  }
}

Modify file app.module.ts

import {Module} from '@nestjs/common'
import {ConfigModule} from '@nestjs/config'
import {ProductsController} from './controller/product.controller'
import {drizzleProvider} from './drizzle-orm'
import {BullMqModule} from './module/bullmq.module'
import {CacheManagerModule} from './module/cache-manager.module'
import {ValkeyModule} from './module/valkey.module'
import {ProductService} from './service/product.service'
import {RefreshAheadCacheService} from './service/refresh-ahead.service'
import {CacheRefreshProcessor} from './worker/cache-refresh.processor'

@Module({
  imports: [
    ConfigModule.forRoot({
      isGlobal: true, 
      envFilePath: '.env', 
    }),
    ValkeyModule,
    BullMqModule,
    CacheManagerModule,
  ],
  controllers: [
    ProductsController,
  ],
  providers: [
    drizzleProvider,
    RefreshAheadCacheService,
    CacheRefreshProcessor,
    ProductService,
  ],
})
export class AppModule {}