Workers - installation

There are two ways to install the psychic-workers package. The first is by selecting yes when prompted during the initial psychic app provisioning stage. If you select yes, the package will be automatically installed, and your app bootstrapped to use workers automatically.

However, if this is not the case for you and you are looking to install websockets after the fact, you can follow these steps:

1. Install the package.

yarn add @rvoh/psychic-workers

2. Add the missing configuration file to src/conf/workers.ts:

import os from 'os'
import { PsychicApplicationWorkers } from '@rvoh/psychic-workers'
import Redis, { Cluster } from 'ioredis'
import AppEnv from './AppEnv'

export default (workersApp: PsychicApplicationWorkers) => {
  workersApp.set('background', {
    defaultWorkstream: {
      // https://docs.bullmq.io/guide/parallelism-and-concurrency
      workerCount: os.cpus().length,
      concurrency: 100,
    },

    defaultQueueConnection: AppEnv.isProduction
      ? new Cluster(
          [
            {
              host: AppEnv.string('BG_JOBS_REDIS_HOST'),
              port: AppEnv.integer('BG_JOBS_REDIS_PORT'),
            },
          ],
          {
            slotsRefreshTimeout: 5000,
            dnsLookup: (address, callback) => callback(null, address),
            redisOptions: {
              username: AppEnv.string('BG_JOBS_REDIS_USERNAME'),
              password: AppEnv.string('BG_JOBS_REDIS_PASSWORD'),
              tls: AppEnv.isProduction ? {} : undefined,
            },
            enableOfflineQueue: false,
          },
        )
      : new Redis({
          host: process.env.BACKGROUND_JOBS_REDIS_HOST || 'localhost',
          port: parseInt(process.env.BACKGROUND_JOBS_REDIS_PORT || '6379', 10),
          username: process.env.BACKGROUND_JOBS_REDIS_USERNAME,
          password: process.env.BACKGROUND_JOBS_REDIS_PASSWORD,
          tls: AppEnv.isProduction ? {} : undefined,
          enableOfflineQueue: false,
        }),

    defaultWorkerConnection: !process.env.WORKER_SERVICE
      ? undefined
      : AppEnv.isProduction
        ? new Cluster(
            [
              {
                host: AppEnv.string('BG_JOBS_REDIS_HOST'),
                port: AppEnv.integer('BG_JOBS_REDIS_PORT'),
              },
            ],
            {
              slotsRefreshTimeout: 5000,
              dnsLookup: (address, callback) => callback(null, address),
              redisOptions: {
                username: AppEnv.string('BG_JOBS_REDIS_USERNAME'),
                password: AppEnv.string('BG_JOBS_REDIS_PASSWORD'),
                tls: AppEnv.isProduction ? {} : undefined,
                maxRetriesPerRequest: null,
              },
            },
          )
        : new Redis({
            host: process.env.BACKGROUND_JOBS_REDIS_HOST || 'localhost',
            port: parseInt(
              process.env.BACKGROUND_JOBS_REDIS_PORT || '6379',
              10,
            ),
            username: process.env.BACKGROUND_JOBS_REDIS_USERNAME,
            password: process.env.BACKGROUND_JOBS_REDIS_PASSWORD,
            tls: AppEnv.isProduction ? {} : undefined,
            maxRetriesPerRequest: null,
          }),
  })

  // ******
  // HOOKS:
  // ******

  workersApp.on('workers:shutdown', () => {
    // add worker shutdown sequence here
  })
}

3. You will need to register this plugin as part of the initializePsychicApplication.ts file within your project. It will need to look something like this:

// initializePsychicApplication.ts

import {
  PsychicApplication,
  PsychicApplicationInitOptions,
} from '@rvoh/psychic'
import { PsychicApplicationWorkers } from '@rvoh/psychic-workers'
import psychicConf from '../../conf/app'
import dreamConf from '../../conf/dream'
import workersConf from '../../conf/workers'

export default async function initializePsychicApplication(
  opts: PsychicApplicationInitOptions = {},
) {
  const psychicApp = await PsychicApplication.init(psychicConf, dreamConf, opts)

  // add this line to integrate psychic workers with your app
  await PsychicApplicationWorkers.init(psychicApp, workersConf)

  return psychicApp
}

4. You will need to add some base classes to your system. These are the classes that end up providing all the backgrounding and scheduling functionality throughout your app, and they are also necessary to bridge types across your application.

// app/services/ApplicationBackgroundedService.ts

import { BaseBackgroundedService } from '@rvoh/psychic-workers'
import psychicTypes from '../../types/psychic'

export default class ApplicationBackgroundedService extends BaseBackgroundedService {
  public get psychicTypes() {
    return psychicTypes
  }
}

// app/services/ApplicationScheduledService.ts

import { BaseScheduledService } from '@rvoh/psychic-workers'
import psychicTypes from '../../types/psychic'

export default class ApplicationScheduledService extends BaseScheduledService {
  public get psychicTypes() {
    return psychicTypes
  }
}

// app/models/ApplicationBackgroundedModel

import {
  BackgroundJobConfig,
  BaseBackgroundedModel,
} from '@rvoh/psychic-workers'
import { DBClass } from '../../types/db'
import { globalSchema, schema } from '../../types/dream'
import psychicTypes from '../../types/psychic'

export default class ApplicationBackgroundedModel extends BaseBackgroundedModel {
  public DB: DBClass

  public static get backgroundJobConfig(): BackgroundJobConfig<BaseBackgroundedModel> {
    return {}
  }

  public get schema() {
    return schema
  }

  public get globalSchema() {
    return globalSchema
  }

  public get psychicTypes() {
    return psychicTypes
  }
}