import { Module } from '@nestjs/common';
import { BullModule } from '@nestjs/bull';
import { ScheduleModule } from '@nestjs/schedule';
import { DigestProducer } from './digest.producer';
import { DigestProcessor } from './digest.processor';
@Module({
imports: [
ScheduleModule.forRoot(),
BullModule.registerQueue({
name: 'email-digest',
defaultJobOptions: {
attempts: 3,
backoff: { type: 'exponential', delay: 5000 },
removeOnComplete: true,
removeOnFail: 1000,
},
}),
],
providers: [DigestProducer, DigestProcessor],
})
export class DigestModule {}
import { Injectable, Logger } from '@nestjs/common';
import { InjectQueue } from '@nestjs/bull';
import { Cron, CronExpression } from '@nestjs/schedule';
import { Queue } from 'bull';
import { UsersService } from '../users/users.service';
@Injectable()
export class DigestProducer {
private readonly logger = new Logger(DigestProducer.name);
constructor(
@InjectQueue('email-digest') private readonly queue: Queue,
private readonly usersService: UsersService,
) {}
@Cron(CronExpression.EVERY_HOUR)
async enqueueDailyDigests(): Promise<void> {
const users = await this.usersService.findDueForDigest(new Date());
const bucket = this.currentHourBucket();
for (const user of users) {
await this.queue.add(
'send-digest',
{ userId: user.id },
{ jobId: `digest:${user.id}:${bucket}` },
);
}
this.logger.log(`Enqueued ${users.length} digest jobs for bucket ${bucket}`);
}
private currentHourBucket(): string {
return new Date().toISOString().slice(0, 13); // YYYY-MM-DDTHH
}
}
import { Logger } from '@nestjs/common';
import {
Process,
Processor,
OnQueueFailed,
OnQueueCompleted,
} from '@nestjs/bull';
import { Job } from 'bull';
import { UsersService } from '../users/users.service';
import { DigestBuilder } from './digest.builder';
import { MailerService } from '../mail/mailer.service';
interface DigestJobData {
userId: string;
}
@Processor('email-digest')
export class DigestProcessor {
private readonly logger = new Logger(DigestProcessor.name);
constructor(
private readonly usersService: UsersService,
private readonly builder: DigestBuilder,
private readonly mailer: MailerService,
) {}
@Process('send-digest')
async handleDigest(job: Job<DigestJobData>): Promise<void> {
const user = await this.usersService.findById(job.data.userId);
if (!user || !user.digestSubscribed) {
return; // nothing to do; do not throw or it will retry
}
const content = await this.builder.buildFor(user);
if (content.isEmpty) {
return;
}
await this.mailer.send({
to: user.email,
subject: content.subject,
html: content.html,
});
}
@OnQueueCompleted()
onCompleted(job: Job<DigestJobData>): void {
this.logger.log(`Digest sent for user ${job.data.userId}`);
}
@OnQueueFailed()
onFailed(job: Job<DigestJobData>, err: Error): void {
this.logger.error(
`Digest job ${job.id} failed (attempt ${job.attemptsMade}): ${err.message}`,
);
}
}
This snippet shows how recurring email digests are scheduled and processed in NestJS using @nestjs/bull, which wraps the Redis-backed Bull queue library. The design separates two concerns: a producer that decides when and for whom a digest should run, and a processor that does the actual work of building and sending each digest. Keeping these apart means the scheduling logic stays thin and the heavy, retry-prone work lives in isolated workers.
In DigestModule, the queue named email-digest is registered with BullModule.registerQueue, along with defaultJobOptions that apply to every job: three attempts with exponential backoff, plus removeOnComplete and a bounded removeOnFail so Redis does not accumulate dead job data indefinitely. Registering options at the queue level avoids repeating them at every add call.
DigestProducer combines a @Cron schedule from @nestjs/schedule with the injected queue. The @Cron handler enqueueDailyDigests runs once an hour, fetches the users due for a digest, and enqueues one job per user. The critical detail is jobId: it is derived from the user id and the current hour bucket, which makes the enqueue idempotent. If the cron fires twice, or two instances run concurrently, Bull deduplicates by jobId and only one job survives — this prevents duplicate emails, a common failure mode in distributed schedulers.
DigestProcessor is decorated with @Processor('email-digest') and marks handleDigest with @Process. It loads the user, guards against a missing or unsubscribed record by returning early (rather than throwing, which would trigger pointless retries), aggregates the digest content, and sends it. Any transient failure — say the mail provider times out — throws, and Bull re-runs the job under the backoff policy defined in the module. The lifecycle hooks @OnQueueFailed and @OnQueueCompleted provide observability without coupling to any specific logging stack.
The trade-off worth noting: idempotency here is keyed on a time bucket, so choosing the bucket granularity (per hour vs per day) determines the dedupe window. A reader would reach for this pattern whenever periodic, per-entity work must run reliably, survive restarts, and never double-fire.
Related snips
require "csv"
class PeopleCsvStream
include Enumerable
HEADERS = %w[id full_name email signed_up_at plan].freezeResilient CSV Export as a Streamed Response
# app/mailers/user_mailer.rb
class UserMailer < ApplicationMailer
default from: 'noreply@example.com'
def welcome_email(user)
@user = userActionMailer advanced patterns for transactional emails
json.array! @posts do |post|
json.cache! ['v1', post], expires_in: 1.hour do
json.id post.id
json.title post.title
json.excerpt post.excerpt
json.published_at post.published_atFragment caching for expensive JSON serialization
import { randomBytes, createHash } from "crypto";
import jwt from "jsonwebtoken";
import { RefreshTokenStore } from "./store";
const ACCESS_SECRET = process.env.ACCESS_SECRET!;
const ACCESS_TTL = "15m";JWT access + refresh token rotation (conceptual)
module EmailNormalization
extend ActiveSupport::Concern
included do
attr_accessor :soft_warningsSoft Validation: Normalize + Validate Email
class CreateTopSellersMv < ActiveRecord::Migration[7.0]
def up
execute <<~SQL
CREATE MATERIALIZED VIEW top_sellers AS
SELECT p.id AS product_id,
p.name AS product_name,Cache-Friendly “Top N” with Materialized View Refresh
Share this code
Here's the card — post it anywhere.
