@asyncview/nestjs

AsyncView SDK for NestJS - Native queue observability for BullMQ and @nestjs/schedule through automatic event capture.

Installation

npm install @asyncview/nestjs

Quick Start

Simply add AsyncViewModule to your app and you're done! AsyncView will automatically discover all your @Processor decorated classes as well as all your @Cron, @Interval, and @Timeout decorated classes and start capturing events.

With BullMQ (@nestjs/bullmq)

import { Module } from '@nestjs/common'
import { BullModule } from '@nestjs/bullmq'
import { ScheduleModule } from '@nestjs/schedule'
import { AsyncViewModule } from '@asyncview/nestjs'

@Module({
  imports: [
    BullModule.forRoot({
      connection: {
        host: 'localhost',
        port: 6379,
      },
    }),
    ScheduleModule.forRoot(),
    // Make sure you import AsyncViewModule after all other modules
    AsyncViewModule.forRoot({
      hubUrl: 'https://your-asyncview-hub.com',
      systemToken: 'asyncview_sys_your_token_here',
    }),
  ],
})
export class AppModule {}

Example BullMQ Processor

import { Processor, WorkerHost } from '@nestjs/bullmq'
import { Job } from 'bullmq'

@Processor('email-queue')
export class EmailProcessor extends WorkerHost {
  async process(job: Job): Promise<void> {
    // Your job processing logic
    // AsyncView automatically captures all job events!
  }
}

With @nestjs/schedule

import { Injectable } from '@nestjs/common'
import { Cron, Interval, Timeout } from '@nestjs/schedule'

@Injectable()
export class TasksService {
  @Cron('0 0 * * *', { name: 'daily-cron' })
  handleDailyCron() {
    // AsyncView automatically tracks execution, duration, and failures
  }

  @Interval(10000, { name: 'interval' })
  handleInterval() {
    // Tracked as scheduler:interval
  }

  @Timeout(5000, { name: 'timeout' })
  handleTimeout() {
    // Tracked as scheduler:timeout
  }
}


That's it! AsyncView will automatically discover and attach to all your processors and schedulers.

## Configuration

```typescript
AsyncViewModule.forRoot({
  hubUrl: string                     // Required: AsyncView hub URL
  systemToken: string                // Required: System-specific token
  captureJobData?: boolean           // Default: true
  captureFailedReason?: boolean      // Default: true
  captureStackTrace?: boolean        // Default: false
  scrubFields?: string[]             // Fields to redact, e.g., ['password', 'ssn']
  sampleRate?: number                // Default: 1.0 (100% of jobs)
  batchSize?: number                 // Default: 100
  flushInterval?: number             // Default: 1000ms
  maxRetries?: number                // Default: 20
  initialRetryDelay?: number         // Default: 1000ms
  maxRetryDelay?: number             // Default: 180000ms (3 minutes)
  retryBackoffMultiplier?: number    // Default: 2
  environment?: string               // Default: process.env.NODE_ENV
  version?: string                   // Default: process.env.npm_package_version
  hostname?: string                  // Default: os.hostname()
})

Privacy & Performance

Data Scrubbing

Protect sensitive information with powerful field scrubbing that supports nested keys, arrays, and wildcards:

AsyncViewModule.forRoot({
  hubUrl: '...',
  systemToken: '...',
  scrubFields: [
    'password',              // Top-level fields
    'user.email',            // Nested fields
    'users[].ssn',           // Fields in arrays
    '*.apiKey',              // Any immediate child's apiKey
    '**.token',              // Any token at any depth
  ],
})

Sampling

For high-throughput queues, reduce overhead by sampling:

AsyncViewModule.forRoot({
  hubUrl: '...',
  systemToken: '...',
  sampleRate: 0.1, // Monitor 10% of jobs
})

Retry Configuration

The SDK includes automatic retry with exponential backoff for hub failures:

AsyncViewModule.forRoot({
  hubUrl: '...',
  systemToken: '...',
  maxRetries: 20,
  initialRetryDelay: 1000,
  maxRetryDelay: 180000,
  retryBackoffMultiplier: 2,
})

Default settings provide ~40 minutes of retry tolerance, ensuring events aren't lost during temporary hub outages.

Queue Library Support

AsyncView automatically detects and supports both queue libraries:

• BullMQ (bullmq + @nestjs/bullmq) - The modern, Redis-based queue library
• @nestjs/schedule (@nestjs/schedule) - The schedule library for cron, interval, and timeout jobs

You only need to install the queue library you're using. AsyncView will automatically:

1. Detect which library is installed
2. Initialize the appropriate adapter
3. Discover and attach to your @Processor decorated classes
4. Discover and attach to your @Cron, @Interval, and @Timeout decorated classes

Events Captured

Queue Jobs (Both BullMQ and Bull)

The SDK automatically captures:

• job.created - When a job is added to the queue
• job.active - When a worker picks up a job
• job.completed - When a job finishes successfully
• job.failed - When a job fails

Scheduled Tasks

The SDK automatically captures @nestjs/schedule decorators:

• schedule.started - When a scheduled task starts execution
• schedule.completed - When a scheduled task completes successfully
• schedule.failed - When a scheduled task fails

Supported decorators:

• @Cron() - Cron-based scheduling
• @Interval() - Interval-based scheduling
• @Timeout() - One-time timeout-based scheduling