JobsModule Overview

The JobsModule is a key component of our Jobs & Workers infrastructure, designed to manage and process a queue of tasks efficiently. This module is crucial for registering tasks that will later be executed by workers. These workers are dedicated to consuming tasks from the queue, ensuring that each task is handled promptly and reliably.

Core Features of JobsModule

Task Registry and Management

At the heart of the JobsModule is its ability to serve as a registry for tasks. This involves creating tasks in a dedicated database table named jobs, which effectively acts as a queue. Each task registered in this queue is intended to be picked up by a worker, which executes the task based on the logic defined for it. This setup allows for a highly organized and scalable approach to managing tasks that need to be executed asynchronously or in the background.

Integration with CloudRunJobs

An advanced feature of the JobsModule is its integration with the CloudRunJobsModule. This integration enables the remote execution of tasks on Google Cloud Run Jobs, allowing tasks to be processed on separate instances. This feature is particularly useful for tasks that are resource-intensive or require isolation from the main application environment. By leveraging Google Cloud Run Jobs, the JobsModule can ensure tasks are handled in the most efficient and scalable manner possible.

Key Benefits

• Scalability: Easily scale your task processing by adding more workers or leveraging cloud-based resources like Google Cloud Run Jobs.
• Reliability: The queue-based system ensures that tasks are processed in order and retries can be easily managed, enhancing the overall reliability of task execution.
• Flexibility: With support for both local workers and Cloud Run Jobs, you have the flexibility to choose the best execution environment for each task.

Ideal Use Cases

• Background Processing: Ideal for applications that require heavy lifting to be done in the background, such as processing large files, generating reports, or performing batch data updates.
• Microservices Workloads: Perfect for microservices architectures where tasks can be distributed across different services and executed independently.
• Resource-Intensive Tasks: Suitable for tasks that are too resource-intensive to run within the main application environment, requiring the capabilities of Google Cloud Run Jobs for execution.

The JobsModule is designed to be easy to integrate and use, making it an essential part of any application that requires efficient task management and execution. Whether you're running simple background jobs or complex, resource-intensive tasks, the JobsModule provides the tools you need to manage and execute these tasks effectively.

Key Components

The module provides a set of reusable components that can be easily integrated into a NestJS project:

Job Entity

The JobEntity is a foundational element within the JobsModule, designed to encapsulate all relevant details about a job within the job management system. This entity is mapped to the jobs database table, holding information about each task queued for execution.

Entity Inheritance

NestKit provide base classes that can be inherited and customized with your needs. For entities we use TypeORM TableInheritance.

Jobs Service

The JobsService is the backbone of the JobsModule, responsible for all interactions with the job queue. This service provides methods for creating new tasks, updating task statuses, and retrieving tasks from the queue. Through the JobsService, applications can programmatically manage their tasks, schedule new jobs, and handle the lifecycle of each task from initiation to completion.

Jobs Resolver

The JobsResolver plays a crucial role in interfacing with GraphQL, making it possible for clients to interact with the JobsModule through GraphQL queries and mutations. It handles requests related to job management, such as submitting new tasks to the queue, fetching the status of existing tasks, and retrieving the outcomes of completed tasks.

Jobs Registry

A central component of the JobsModule is the Jobs Registry, a comprehensive listing of all task types that the module can handle. The registry allows developers to define and register different types of jobs, specifying the worker where job should be prcessed. This component ensures that when a task is added to the queue, the system knows exactly how to handle and execute it, based on its registered type.

CloudRun Jobs Module

Integrating the CloudRunJobsModule with the JobsModule is straightforward, involving the registration of jobs as Cloud Run tasks and configuring the CloudRunJobsService to handle these tasks. This integration enables developers to easily specify which jobs should be processed on Cloud Run, providing a flexible and powerful solution for job execution.

How to install Jobs Module to my app

To create a JobsModule in NestJS, you can follow these steps:

1. Install the required dependencies for JobsModule:

   • Id Module
   • Winston Module
   • CloudRun Jobs Module

2. Create a jobs folder in your project's src/modules directory.

3. Inside the jobs/types folder, create an common.ts file where you will define WorkerType enum:

   export enum WorkerType {
     DIGGER = 'DIGGER',
     COPTER = 'COPTER',
     AIRPLANE = 'AIRPLANE',
   }

4. Inside the jobs/registry folder, create an test-job.entity.ts file where you will define your first job entity:

   import { Job } from '@deeepvision/nest-kit/dist/modules/jobs';
   import { ObjectType } from '@nestjs/graphql';
   import { ChildEntity } from 'typeorm';
   
   @ObjectType()
   export class TestJobMeta {
     @Field()
     storage!: StorageType;
   }
   
   @ObjectType({
     implements: () => Job,
   })
   @ChildEntity()
   export class TestJob extends Job {
     @Field(() => TestJobMeta)
     meta!: TestJobMeta;
   }

5. Inside the jobs/registry folder, create an index.ts file where you will store all your job types and worker mappings:

   import { JobRegistry, ScheduledJobRegistryItem } from '@/modules/jobs';
   import { SendNotificationsJob } from '@/modules/notifications';
   import { withDashEnv } from '@/tools';
   
   import { WorkerType } from '../types/common';
   import { GenerateWaveformJob } from './generate-waveform-job.entity';
   import { TestJob } from './test-job.entity';
   
   /**
    * Registry of job types.
    */
   export const jobRegistry: JobRegistry = {
     [TestJob.name]: TestJob,
   };
   
   /**
   * Mapping of job types to worker types.
   */
   export const jobTypeToWorkerType: Record<string, string> = {
     [TestJob.name]: WorkerType.COPTER,
   };
   
   /**
   * Mapping of worker types to Cloud Run Job names where jobs should be executed.
   */
   export const workerTypeToCloudRunJobName: Record<string, string> = {
     [WorkerType.COPTER]: `my-copter-worker`,
   };
   
   /**
   * Registry of scheduled jobs.
   */
   export const scheduledJobRegistry: ScheduledJobRegistryItem[] = [];

6. Inside the jobs folder, create a jobs.service.ts file and extends BaseJobsService from NestKit with the following code:

   // src/modules/jobs/jobs.service.ts
   
   import {
     BaseJobsService, CreateJobOptions, Job,
   } from '@deeepvision/nest-kit/dist/modules/jobs';
   import { Injectable, Type } from '@nestjs/common';
   import { ServiceMethodContext } from '@/types/services';
   import {
     jobRegistry, jobTypeToWorkerType, scheduledJobRegistry, workerTypeToCloudRunJobName,
   } from './registry';
   import { TestJob, TestJobMeta } from './registry/test-job.entity';
   
   @Injectable()
   export class JobsService extends BaseJobsService {
     constructor() {
       super(
         jobRegistry,
         jobTypeToWorkerType,
         workerTypeToCloudRunJobName,
         scheduledJobRegistry,
       );
     }
   
     /**
      * Create a new TestJob.
      * For each job type, you should create a method like this.
      */
     async createTestJob(opts: CreateJobOptions<TestJobMeta>, ctx: ServiceMethodContext) {
       return await this.create(TestJob, opts, ctx);
     }
   }

7. Inside the jobs folder, create a jobs.resolver.ts file and extends BaseJobsResolver from NestKit with the following code:

   // src/modules/jobs/jobs.resolver.ts
   import { BaseJobsResolver, Job } from '@deeepvision/nest-kit/dist/modules/jobs';
   import { Resolver } from '@nestjs/graphql';
   
   @Resolver(() => Job)
   export class JobsResolver extends BaseJobsResolver(Job) {
     constructor() {
       super();
     }
   }

8. Inside the jobs folder, create a jobs.module.ts file with the following code:

   // src/modules/jobs/jobs.module.ts
   import { CloudRunJobsModule, JOBS_SERVICE_TOKEN } from '@deeepvision/nest-kit/dist/modules/jobs';
   import { Module } from '@nestjs/common';
   
   import { JobsController } from './jobs.controller';
   import { JobsResolver } from './jobs.resolver';
   import { JobsService } from './jobs.service';
   
   @Module({
     imports: [
       CloudRunJobsModule,
     ],
     providers: [
       JobsService,
       {
         provide: JOBS_SERVICE_TOKEN,
         useExisting: JobsService,
       },
       JobsResolver,
     ],
     controllers: [
       JobsController,
     ],
     exports: [
       JobsService,
       {
         provide: JOBS_SERVICE_TOKEN,
         useExisting: JobsService,
       },
     ],
   })
   export class JobsModule {}

9. Finally, import the JobsModule into your main AppModule or another module:

   // src/modules/app.module.ts
   
   import { Module } from '@nestjs/common';
   import { JobsModule } from './modules/jobs/jobs.module.ts';
   
   @Module({
     imports: [JobsModule],
   })
   export class AppModule {}

How to create new job

To create a new job, you need to inject the JobsService into your service and call the appropriate method to create the job. Here's an example of how to create a new TestJob:

import { JobsService } from '@/modules/jobs';
import { TestJobMeta } from '@/modules/jobs/registry/test-job.entity';

@Injectable()
export class ProcessingService {
  constructor(
    private readonly jobsService: JobsService,
  ) {}

  async runTestProcessing(myDocumentId: string, ctx: ServiceMethodContext) {
    await this.jobsService.createTestJob({
      referenceId: myDocumentId,
      meta: {
        storage: StorageType.S3,
      },
    }, ctx);
  }
}

How to check job status

Use GraphQL to query the job status. Here's an example of how to query the job by its ID: