add pgboss client class

Author: levinmr

index.js

@@ -6,6 +6,7 @@ const AppConfig = require("./src/app_config");
 const ReportProcessingContext = require("./src/report_processing_context");
 const Logger = require("./src/logger");
 const Processor = require("./src/processor");
+const PgBossClient = require("./src/pg_boss_client");
 
 /**
  * Gets an array of JSON report objects from the application confing, then runs
@@ -127,7 +128,6 @@ async function runQueuePublish(options = {}) {
     connection: appConfig.postgres,
   });
   const queueClient = await _initQueueClient(knexInstance, appLogger);
-  const queue = "analytics-reporter-job-queue";
 
   for (const agency of agencies) {
     for (const reportConfig of reportConfigs) {
@@ -139,7 +139,7 @@ async function runQueuePublish(options = {}) {
       });
       try {
         let jobId = await queueClient.send(
-          queue,
+          appConfig.messageQueueName,
           _createQueueMessage(
             options,
             agency,
@@ -156,13 +156,17 @@ async function runQueuePublish(options = {}) {
         );
         if (jobId) {
           reportLogger.info(
-            `Created job in queue: ${queue} with job ID: ${jobId}`,
+            `Created job in queue: ${appConfig.messageQueueName} with job ID: ${jobId}`,
           );
         } else {
-          reportLogger.info(`Found a duplicate job in queue: ${queue}`);
+          reportLogger.info(
+            `Found a duplicate job in queue: ${appConfig.messageQueueName}`,
+          );
         }
       } catch (e) {
-        reportLogger.error(`Error sending to queue: ${queue}`);
+        reportLogger.error(
+          `Error sending to queue: ${appConfig.messageQueueName}`,
+        );
         reportLogger.error(util.inspect(e));
       }
     }
@@ -199,7 +203,7 @@ function _initAgencies(agencies_file) {
 async function _initQueueClient(knexInstance, logger) {
   let queueClient;
   try {
-    queueClient = new PgBoss({ db: knexInstance.client.acquireConnection() });
+    queueClient = new PgBoss({ db: new PgBossClient(knexInstance) });
     await queueClient.start();
     logger.debug("Starting queue client");
   } catch (e) {
@@ -242,7 +246,6 @@ async function runQueueConsume() {
     connection: appConfig.postgres,
   });
   const queueClient = await _initQueueClient(knexInstance, appLogger);
-  const queue = "analytics-reporter-job-queue";
 
   try {
     const context = new ReportProcessingContext(new AsyncLocalStorage());
@@ -253,7 +256,7 @@ async function runQueueConsume() {
     );
 
     await queueClient.work(
-      queue,
+      appConfig.messageQueueName,
       { newJobCheckIntervalSeconds: 1 },
       async (message) => {
         appLogger.info("Queue message received");

src/pg_boss_client.js

@@ -0,0 +1,27 @@
+/**
+ * Handles providing a database client for the Pg-Boss library
+ */
+class PgBossClient {
+  #knex;
+
+  /**
+   * @param {import('knex')} knexInstance an initialized instance of the knex
+   * library which provides a database connection.
+   */
+  constructor(knexInstance) {
+    this.#knex = knexInstance;
+  }
+
+  /**
+   * Execute PgBoss SQL using the knex library interface
+   *
+   * @param {string} text the SQL string to execute.
+   * @param {string[]} values the values to insert into the SQL string.
+   * @returns {Promise} which resolves with the result of the SQL query.
+   */
+  async executeSql(text, values) {
+    return this.#knex.raw(text, values);
+  }
+}
+
+module.exports = PgBossClient;