medusa-store/packages/modules/workflow-engine-redis/src/loaders/redis.ts

import {
  InternalModuleDeclaration,
  LoaderOptions,
} from "@medusajs/framework/types"
import { asValue } from "@medusajs/framework/awilix"
import Redis from "ioredis"
import { RedisWorkflowsOptions } from "../types"

export default async (
  { container, logger, options, dataLoaderOnly }: LoaderOptions,
  moduleDeclaration: InternalModuleDeclaration
): Promise<void> => {
  const {
    url,
    options: redisOptions,
    jobQueueName,
    queueName,
    pubsub,
  } = options?.redis as RedisWorkflowsOptions

  // TODO: get default from ENV VAR
  if (!url) {
    throw Error(
      "No `redis.url` provided in `workflowOrchestrator` module options. It is required for the Workflow Orchestrator Redis."
    )
  }

  const cnnPubSub = pubsub ?? { url, options: redisOptions }

  const queueName_ = queueName ?? "medusa-workflows"
  const jobQueueName_ = jobQueueName ?? "medusa-workflows-jobs"

  let connection
  let redisPublisher
  let redisSubscriber
  let workerConnection

  try {
    connection = await getConnection(url, redisOptions)
    workerConnection = await getConnection(url, {
      ...(redisOptions ?? {}),
      maxRetriesPerRequest: null,
    })
    logger?.info(
      `[Workflow-engine-redis] Connection to Redis in module 'workflow-engine-redis' established`
    )
  } catch (err) {
    logger?.error(
      `[Workflow-engine-redis] An error occurred while connecting to Redis in module 'workflow-engine-redis': ${err}`
    )
  }

  try {
    redisPublisher = await getConnection(cnnPubSub.url, cnnPubSub.options)
    redisSubscriber = await getConnection(cnnPubSub.url, cnnPubSub.options)
    logger?.info(
      `[Workflow-engine-redis] Connection to Redis PubSub in module 'workflow-engine-redis' established`
    )
  } catch (err) {
    logger?.error(
      `[Workflow-engine-redis] An error occurred while connecting to Redis PubSub in module 'workflow-engine-redis': ${err}`
    )
  }

  container.register({
    isWorkerMode: asValue(moduleDeclaration.worker_mode !== "server"),
    partialLoading: asValue(true),
    redisConnection: asValue(connection),
    redisWorkerConnection: asValue(workerConnection),
    redisPublisher: asValue(redisPublisher),
    redisSubscriber: asValue(redisSubscriber),
    redisQueueName: asValue(queueName_),
    redisJobQueueName: asValue(jobQueueName_),
    redisDisconnectHandler: asValue(async () => {
      connection.disconnect()
      workerConnection.disconnect()
      redisPublisher.disconnect()
      redisSubscriber.disconnect()
    }),
  })
}

async function getConnection(url, redisOptions) {
  const connection = new Redis(url, {
    lazyConnect: true,
    ...(redisOptions ?? {}),
  })

  await new Promise(async (resolve) => {
    await connection.connect(resolve)
  })

  return connection
}