add debounce support on insert() via singletonSeconds (#485)

Author: timgit

package-lock.json

@@ -1,6 +1,6 @@
 {
   "name": "pg-boss",
-  "version": "10.0.6",
+  "version": "10.1.0",
   "description": "Queueing jobs in Postgres from Node.js like a boss",
   "main": "./src/index.js",
   "engines": {

package.json

@@ -408,7 +408,9 @@ class Manager extends EventEmitter {
       this.config.retryBackoff // 6
     ]
 
-    return await db.executeSql(this.insertJobsCommand, params)
+    const { rows } = await db.executeSql(this.insertJobsCommand, params)
+
+    return (rows.length) ? rows.map(i => i.id) : null
   }
 
   getDebounceStartAfter (singletonSeconds, clockOffset) {

src/manager.js

@@ -728,6 +728,7 @@ function insertJobs (schema) {
       priority,
       start_after,
       singleton_key,
+      singleton_on,
       dead_letter,
       expire_in,
       keep_until,
@@ -743,6 +744,10 @@ function insertJobs (schema) {
       COALESCE(priority, 0) as priority,
       j.start_after,
       "singletonKey" as singleton_key,
+      CASE
+        WHEN "singletonSeconds" IS NOT NULL THEN 'epoch'::timestamp + '1 second'::interval * ("singletonSeconds" * floor( date_part('epoch', now()) / "singletonSeconds" ))
+        ELSE NULL
+        END as singleton_on,
       COALESCE("deadLetter", q.dead_letter) as dead_letter,
       CASE
         WHEN "expireInSeconds" IS NOT NULL THEN "expireInSeconds" *  interval '1s'
@@ -778,7 +783,7 @@ function insertJobs (schema) {
         "retryDelay" integer,
         "retryBackoff" boolean,
         "singletonKey" text,
-        "singletonOn" text,
+        "singletonSeconds" integer,
         "expireInSeconds" integer,
         "keepUntil" timestamp with time zone,
         "deadLetter" text

src/plans.js

@@ -60,7 +60,7 @@ class Timekeeper extends EventEmitter {
       batchSize: 50
     }
 
-    await this.manager.work(QUEUES.SEND_IT, options, (jobs) => this.manager.insert(jobs.map(i => i.data)))
+    await this.manager.work(QUEUES.SEND_IT, options, async (jobs) => { await this.manager.insert(jobs.map(i => i.data)) })
 
     setImmediate(() => this.onCron())
 

src/timekeeper.js

@@ -207,6 +207,7 @@ declare namespace PgBoss {
     retryBackoff?: boolean;
     startAfter?: Date | string;
     singletonKey?: string;
+    singletonSeconds?: number;
     expireInSeconds?: number;
     keepUntil?: Date | string;
     deadLetter?: string;