WRITE-BEHIND

Introduction

This article will guide you through implementing WRITE-BEHIND (WRITE-BACK) in NestJS with Redis and Postgres.
Applying this mechanism will perform asynchronous writes to the Database via a Queue, so I will also use @nestjs/bullmq (BullMQ + Redis), which is a standardized, powerful library for managing background jobs.
As for theoretical content, you can review the previous article I mentioned.

Detail

Please install the following packages:

bun add bullmq @nestjs/bullmq

Create Drizzle Schema in drizzle-orm/schema/schema.ts as follows:

import {
  bigserial,
  decimal,
  integer,
  pgTable,
  serial,
  text,
  timestamp,
  varchar,
} from 'drizzle-orm/pg-core'

export const products = pgTable('products', {
  id: bigserial('id', {mode: 'number'}).primaryKey(),
  name: varchar('name', {length: 255}).notNull(),
  description: text('description'),
  price: decimal('price', {precision: 10, scale: 2}).notNull(),
  views: integer('views').default(0),
  createdAt: timestamp('created_at').defaultNow().notNull(),
  updatedAt: timestamp('updated_at').defaultNow().notNull(),
})

Create Queue and Job to use in BullMQ constant/index.ts:

export enum QueueName {
  PRODUCT_VIEWS = 'product-views',
}

export enum JobName {
  SYNC_VIEW_COUNT = 'sync-view-count',
}

Create file module/bullmq.module.ts to connect with Redis and register a queue:

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.PRODUCT_VIEWS,
    }),
  ],
  exports: [
    BullModule,
  ],
})
export class BullMqModule {}

Create file worker/product-view.processor.ts:

import {Processor, WorkerHost} from '@nestjs/bullmq'
import {Inject, Injectable} from '@nestjs/common'
import {Job} from 'bullmq'
import {eq} from 'drizzle-orm'
import {NodePgDatabase} from 'drizzle-orm/node-postgres'
import {JobName, QueueName} from 'src/constant'
import * as schema from 'src/drizzle-orm/schema/schema'

@Processor(QueueName.PRODUCT_VIEWS)
@Injectable()
export class ProductViewProcessor extends WorkerHost {
  constructor(
    @Inject('DRIZZLE_DB') private readonly db: NodePgDatabase<typeof schema>
  ) {
    super()
  }

  async process(job: Job<{productId: number; views: number}, any, string>) {
    switch (job.name) {
      case JobName.SYNC_VIEW_COUNT: {
        const {productId, views} = job.data

        const [updatedProduct] = await this.db
          .update(schema.products)
          .set({
            views, 
            updatedAt: new Date(),
          })
          .where(eq(schema.products.id, productId))
          .returning()

        return updatedProduct
      }
    }
  }
}

You can see that the Processor section handles the Queue, while the process function runs in the background depending on the job name.
This job only performs the update of product views into the Database.

Create file service/product.service.ts:

import {InjectQueue} from '@nestjs/bullmq'
import {Inject, Injectable} from '@nestjs/common'
import {Queue} from 'bullmq'
import {eq} from 'drizzle-orm'
import type {NodePgDatabase} from 'drizzle-orm/node-postgres'
import Redis from 'ioredis'
import {JobName, QueueName} from 'src/constant'
import * as schema from 'src/drizzle-orm/schema/schema'

@Injectable()
export class ProductService {
  constructor(
    @Inject('VALKEY_CLIENT') private readonly valkey: Redis,
    @InjectQueue(QueueName.PRODUCT_VIEWS) private dbQueue: Queue, 
    @Inject('DRIZZLE_DB') private readonly db: NodePgDatabase<typeof schema>
  ) {}

  async getProductById(id: number) {
    return await this.db
      .select()
      .from(schema.products)
      .where(eq(schema.products.id, id))
      .then(res => res[0])
  }

  async updateViewCount(productId: number) {
    const cacheKey = `product:views:${productId}`

    const newViews = await this.valkey.incr(cacheKey)

    await this.dbQueue.add(
      JobName.SYNC_VIEW_COUNT,
      {
        productId,
        views: newViews,
      },
      {
        attempts: 3,
        backoff: 5000,
      }
    )

    return {success: true, views: newViews}
  }
}

The updateViewCount function performs WRITE-BEHIND (WRITE-BACK) by directly incrementing the views in Redis first, then pushing the job to the Queue for background processing to update product views in the Database.
The attempts: 3 setting retries 3 times if an error occurs and backoff: 5000 sets a 5-second interval between retries.

Create file controller/product.controller.ts:

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

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

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

  @Patch(':id/view')
  async incrementView(@Param('id', ParseIntPipe) id: number) {
    return await this.productService.updateViewCount(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 {ValkeyModule} from './module/valkey.module'
import {ProductService} from './service/product.service'
import {ProductViewProcessor} from './worker/product-view.processor'

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

As a result, you can see that when using products/1/views, the processing time is extremely fast because it only needs to write to Redis, while updating the Database is processed in the background so there is no extra wait time.
Note that this should only be used for non-critical data where minor discrepancies are acceptable, such as view counts, because if a job fails to run, data loss can easily occur.