How do I retrieve a queue created in another file to add a job? (Without polluting global namespace)

[stephanoparaskeva]

Incorrect:

const queue = new Queue('PAYMENTS_QUEUE', { connection: { host: 'localhost', port: 6379 } });
export default queue

Desired

(() => {
  new Queue('PAYMENTS_QUEUE', { connection: { host: 'localhost', port: 6379 } });
})()

(() => {
  const queue = getQueue(PAYMENTS_QUEUE)
  queue.add('MAKE_PAYMENT', { to: 1 })
})()

[manast]

I am not sure what your issue is, you just use the same name of the queue that you want to refer to.

[gabegorelick]

Is the issue that instantiating a queue object has side effects (like creating connections to Redis) so you don't want that to happen when the module is imported?

The way I've dealt with that in the past is with something like this:

let queue;

module.exports = () => {
  if (!queue) {  queue = new Queue(); }
  return queue;
};

Then you're not polluting the global namespace, just your module-local one.

[stephanoparaskeva]

The thing I'm trying to do, is instead of instantiating the new Queue and importing it every where I need it, can't I just instantiate it, and refer to it by a string?

Currently my app works like this:

const setupQueues = () => {
  const queue = new Queue(PAYMENTS_QUEUE, { connection: { host: 'localhost', port: 6379 } });
  const queueScheduler = new QueueScheduler(PAYMENTS_QUEUE);
  const queueEvents = new QueueEvents(PAYMENTS_QUEUE);

  return () => ({ queue, queueEvents, queueScheduler });
};

export default setupQueues;

// App.tsx

// sets up queues
export const queues = setupQueues();
paymentSystem();

// paymentSystem

const paymentSystem = async () => {
  // These are necessary to add Workers to work on the future delayed MAKE_PAYMENT task
  const { queueEvents, queueScheduler } = queues();
  await queueScheduler.waitUntilReady();
  await queueEvents.waitUntilReady();

  // This runs the callback when the job is ready to be processed after the delay has passed
  new Worker<CallEvent>(PAYMENTS_QUEUE, async job => {
    const { callFromId, callToId, callId, id } = job.data;
....

// CallEvent.ts

export const saveNewCallEvent = async (jwtPayload: any, data: TCreateCallEventData) => {
  // Checks that at least of the users in the callEvent is the user making the request
  if (!isGivenUser(jwtPayload, data.callFromId) && !isGivenUser(jwtPayload, data.callToId)) {
    throw createHttpError(401, 'Unauthorized');
  }

  try {
    // Ensures that the callEventCreatedById is the actual ID of the given jwtPayload from the request
    const callEvent = await callEventRepo().save({ ...data, callEventCreatedById: jwtPayload.sub });
    const { queue } = queues();
    queue.add(MAKE_PAYMENT, callEvent, { removeOnComplete: true, delay: ONE_MINUTE / 2 }); // Todo Change delay to 30 minutes (from 30 seconds)
  } catch (err) {
    throw isDuplicateKeyErr(err) ? createHttpError(403, { err }) : err;
  }
};

// server.ts

const server = createServer(app);

const connect = () => createConnection(connectionOptions());
const listen = async () =>
  server.listen(port, host, () => console.log(`on http://${host}:${port}`));

const attemptStart = (retries = 10) =>
  healthReq()
    .then(connect)
    .then(listen)
    .catch(async err => {
      console.log({ retries, err });
      if (retries) wait(10000).then(() => attemptStart(retries - 1));
    });

attemptStart();

Surely there's a better way to do this?

[manast]

this code should not be necessary:

  // These are necessary to add Workers to work on the future delayed MAKE_PAYMENT task
  const { queueEvents, queueScheduler } = queues();
  await queueScheduler.waitUntilReady();
  await queueEvents.waitUntilReady();

Also do not forget to use await when adding jobs or you will miss exceptions:

    await queue.add(MAKE_PAYMENT, callEvent, { removeOnComplete: true, delay: ONE_MINUTE / 2 }); // Todo Change delay to 30 minutes (from 30 seconds)

[stephanoparaskeva]

Thank you for overviewing that, that could have caught me out. If you have a moment, perhaps you can take a look at this payment system I built with bull

// payment_system.ts

import { getRepository } from 'typeorm';
import { Worker } from 'bullmq';
import { queues } from '../../app';
import { PAYMENTS_QUEUE } from './queues';
import { CallEvent } from './call_events/call_event.entity';
import { ONE_MINUTE, ONE_SECOND } from '../../meta/constants/numeric_consts';
import { createFailedPayment, createSuccessfulPayment } from './payments/payment.service';

const pricePerMinute = 1; // Todo change to user value

const DISCREPANCY_ERROR = 'discrepancy between both users timeline of events';
const SHORT_CALL_ERROR = 'call too short, no payment processed';

// General Flow:
// Event 'post request' comes in, and gets saved to DB.
// A Job for that event is created with a 30 minute delay.
// The Worker then processes the Job after 30 minutes have passed.
// The Worker checks if there has been an event posted to DB since. (Within this 30 minute period).
// If there has been, return and do nothing. IE wait. The next event's job will take care of the rest.
// If there hasn't been any event since and if this events' state is either ‘disconnecting’ || ‘failure’ || ‘closed’.
// We finally make the payment.
const paymentSystem = async () => {
  // These are necessary to add Workers to work on the future delayed MAKE_PAYMENT task
  const { queueEvents, queueScheduler } = queues();
  await queueScheduler.waitUntilReady();
  await queueEvents.waitUntilReady();

  // This runs the callback when the job is ready to be processed after the delay has passed
  new Worker<CallEvent>(PAYMENTS_QUEUE, async job => {
    const { callFromId, callToId, callId, id } = job.data;

    try {
      // Gets all events of call, to find most recent.
      // This is to check if the current job is dealing with the most recent call event.
      const callEvents: CallEvent[] | [] = await getRepository(
        CallEvent
      ).query(`select * from call_events where "callId" = $1`, [callId]);

      const initiatorEvents = callEvents.filter(
        event => event.callEventCreatedById === event.callFromId
      );
      const recipientEvents = callEvents.filter(
        event => event.callEventCreatedById !== event.callFromId
      );
      // This says that both users agree there was an ongoing call, with a start and end
      // This is the core protective measure, a user could potentially create callEvents manually
      // Switching over the callToId and callFromId, there is no safeguard against this,
      // But the callEventCreatedById is tightly controlled, and so only a user with a given JWT
      // Can create events that say they are createdBy that person. And so these two arrays should
      // Only both have more than 2 events if 2 separate tokens were used to create those events
      if (initiatorEvents?.length < 2 || recipientEvents?.length < 2) {
        console.log('WAIT, not yet 2 events for both');
        // WAIT: not yet 2 events for both
        return;
      }
      const lastCallEvent = callEvents[callEvents.length - 1];
      // If the current job is not dealing with the most recent call event, return, do nothing.
      // A more recent job must already exist.
      if (id !== lastCallEvent.id) {
        console.log('WAIT, not final event');
        // WAIT: not final event
        return;
      }

      // LOGIC HERE ON IS TO CREATE A PAYMENT

      const firstInitiatorEvent = initiatorEvents[0];
      const lastInitiatorEvent = initiatorEvents[initiatorEvents.length - 1];

      const firstRecipientEvent = recipientEvents[0];
      const lastRecipientEvent = recipientEvents[recipientEvents.length - 1];

      // Otherwise:
      // If this is the most recent event, 30 minutes must therefore have passed. Since that request was made. So:
      // Make The Payment, the call session is therefore finished.
      const initiatorTimeDifferenceMS = Math.abs(
        +new Date(lastInitiatorEvent.createdAt) - +new Date(firstInitiatorEvent.createdAt)
      );
      const recipientTimeDifferenceMS = Math.abs(
        +new Date(lastRecipientEvent.createdAt) - +new Date(firstRecipientEvent.createdAt)
      );
      // If we have remarkably different time-difference accounts, prevent reaching payment
      if (Math.abs(initiatorTimeDifferenceMS - recipientTimeDifferenceMS) >= ONE_MINUTE * 15) {
        // Payment Invalidation: saves payment as failed with an error
        // so no future payment can be saved with this callId (Primary Key)
        await createFailedPayment({ callId, callFromId, callToId, errStr: DISCREPANCY_ERROR });
        return;
      }
      const meanTimeDifferenceMS = (initiatorTimeDifferenceMS + recipientTimeDifferenceMS) / 2;
      const totalCost = (meanTimeDifferenceMS / ONE_MINUTE) * pricePerMinute;

      // TODO change to 3 minutes
      if (meanTimeDifferenceMS < 50 * ONE_SECOND) {
        // Payment Invalidation: saves payment as failed with an error
        // so no future payment can be saved with this callId (Primary Key)
        await createFailedPayment({ callId, callFromId, callToId, errStr: SHORT_CALL_ERROR });
        return;
      }

      await createSuccessfulPayment({
        callId,
        callFromId,
        callToId,
        totalCost,
        durationMS: String(meanTimeDifferenceMS),
      });
    } catch (err) {
      console.warn('General Error', err);
    }
  });
};

export default paymentSystem;

// payment.service.ts

import { getRepository } from 'typeorm';
import { Payment } from './payment.entity';

type TPayment = {
  callId: string;
  callFromId: string;
  callToId: string;
  errStr?: string;
  totalCost?: number;
  durationMS?: string | null;
};

export const createSuccessfulPayment = async ({
  callId,
  callFromId,
  callToId,
  totalCost,
  durationMS,
}: TPayment) => {
  try {
    const payment: [Payment] | [] = await getRepository(
      Payment
    ).query(
      `insert into payments (id, "paymentFromId", "paymentToId", total, "isSuccess", error, "durationMS") values ($1, $2, $3, $4, $5, $6, $7) returning *`,
      [callId, callFromId, callToId, totalCost, true, null, durationMS]
    );
    console.log('Successful Payment Created', payment);
    // todo Make Payment via payment gateway here
    return payment;
  } catch (err) {
    console.warn('Payment Create Error:', err);
    // todo potential Job retry logic here
    return null;
  }
};

export const createFailedPayment = async ({ callId, callFromId, callToId, errStr }: TPayment) => {
  try {
    const payment: [Payment] | [] = await getRepository(Payment).query(
      `insert into payments (id, "paymentFromId", "paymentToId", total, "isSuccess", error, "durationMS")
                 values ($1, $2, $3, $4, $5, $6, $7)
                 returning *`,
      [callId, callFromId, callToId, null, false, errStr, null]
    );
    console.log('Failed Payment Created', payment);
    return payment;
  } catch (err) {
    console.warn('Payment Create Error:', err);
    // todo potential retry logic here
    return null;
  }
};