Skip to main content

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.
pnpm add @rvoh/psychic-workers
  1. Add the missing configuration file to conf/initializers/workers.ts:
// conf/initializers/workers.ts

import { PsychicApp } from '@rvoh/psychic'
import { PsychicAppWorkers } from '@rvoh/psychic-workers'
import { Queue, Worker } from 'bullmq'
import Redis, { Cluster } from 'ioredis'
import AppEnv from '../AppEnv.js'

export default (psy: PsychicApp) => {
  psy.plugin(async () => {
    await PsychicAppWorkers.init(psy, initializeWorkers)
  })
}

function initializeWorkers(workersApp: PsychicAppWorkers) {
  workersApp.set('background', {
    defaultWorkstream: {
      workerCount: 1,
      concurrency: 10,
    },

    namedWorkstreams: [
      {
        name: 'NamedWorkstream',
        workerCount: 1,
      },

      // Rate limited workstream
      // {
      //   name: 'RateLimitedWorkstream',
      //   workerCount: 1,
      //   concurrency: 10,
      //   rateLimit: {
      //     max: 20,
      //     duration: 1000,
      //   },
      // },
    ],

    providers: {
      Queue,
      Worker,
    },

    defaultBullMQQueueOptions: {
      defaultJobOptions: {
        removeOnComplete: 1000,
        removeOnFail: 20000,
        // 524,288,000 ms (~6.1 days) using algorithm:
        // "2 ^ (attempts - 1) * delay"
        attempts: 20,
        backoff: {
          type: 'exponential',
          delay: 1000,
        },
      },
    },

    // Any instance can push onto the queue. This producer (non-blocking) connection
    // sets `enableOfflineQueue: false` so `queue.add()` fails fast when Redis is down
    // instead of buffering jobs in memory that vanish on restart. BullMQ recommends
    // disabling the offline queue on the Queue while leaving it on for Workers.
    // https://docs.bullmq.io/patterns/failing-fast-when-redis-is-down
    defaultQueueConnection: AppEnv.isProduction
      ? new Cluster(
          [
            {
              host: AppEnv.string('BG_JOBS_REDIS_HOST'),
              port: AppEnv.integer('BG_JOBS_REDIS_PORT', { optional: true }) || 6379,
            },
          ],
          {
            slotsRefreshTimeout: 10000,
            dnsLookup: (address, callback) => callback(null, address),
            redisOptions: {
              username: AppEnv.string('BG_JOBS_REDIS_USERNAME'),
              password: AppEnv.string('BG_JOBS_REDIS_PASSWORD'),
              tls: {},
            },
            clusterRetryStrategy: (times: number) => Math.max(Math.min(Math.exp(times), 20000), 1000),
            enableOfflineQueue: false,
          }
        )
      : new Redis({
          host: AppEnv.string('BG_JOBS_REDIS_HOST', { optional: true }) || 'localhost',
          port: AppEnv.integer('BG_JOBS_REDIS_PORT', { optional: true }) || 6379,
          username: AppEnv.string('BG_JOBS_REDIS_USERNAME', { optional: true }),
          password: AppEnv.string('BG_JOBS_REDIS_PASSWORD', { optional: true }),
          // tls:  {},
          retryStrategy: (times: number) => Math.max(Math.min(Math.exp(times), 20000), 1000),
          enableOfflineQueue: false,
        }),

    // Only establish the worker Redis connection if on an instance that does the work.
    // This consumer (blocking) connection sets `maxRetriesPerRequest: null` — required by
    // BullMQ, whose Worker/QueueEvents use blocking commands (BLPOP/BRPOPLPUSH) on a
    // duplicated connection and throw unless it is null. Do not change it, and do NOT copy
    // `null` to non-blocking connections (the queue above, the websockets adapter), which
    // should fail fast. https://docs.bullmq.io/guide/connections
    defaultWorkerConnection: !AppEnv.boolean('WORKER_SERVICE')
      ? undefined
      : AppEnv.isProduction
      ? new Cluster(
          [
            {
              host: AppEnv.string('BG_JOBS_REDIS_HOST'),
              port: AppEnv.integer('BG_JOBS_REDIS_PORT', { optional: true }) || 6379,
            },
          ],
          {
            slotsRefreshTimeout: 15000,
            dnsLookup: (address, callback) => callback(null, address),
            redisOptions: {
              username: AppEnv.string('BG_JOBS_REDIS_USERNAME'),
              password: AppEnv.string('BG_JOBS_REDIS_PASSWORD'),
              tls: {},
              maxRetriesPerRequest: null,
            },
            clusterRetryStrategy: (times: number) => Math.max(Math.min(Math.exp(times), 20000), 1000),
          }
        )
      : new Redis({
          host: AppEnv.string('BG_JOBS_REDIS_HOST', { optional: true }) || 'localhost',
          port: AppEnv.integer('BG_JOBS_REDIS_PORT', { optional: true }) || 6379,
          username: AppEnv.string('BG_JOBS_REDIS_USERNAME', { optional: true }),
          password: AppEnv.string('BG_JOBS_REDIS_PASSWORD', { optional: true }),
          // tls:  {},
          maxRetriesPerRequest: null,
          retryStrategy: (times: number) => Math.max(Math.min(Math.exp(times), 20000), 1000),
        }),
  })

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

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

  1. Nothing further is needed to register the plugin. conf/app.ts calls psy.load('initializers', …), which auto-loads every file in conf/initializers/, including workers.ts above — its default export calls psy.plugin(...), which is what wires PsychicAppWorkers.init(...) into app boot. This is the same plugin-registration mechanism used elsewhere in Psychic; no changes to initializePsychicApp.ts are required.

  2. 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
  }
}