added defer in throttle

Author: ackava

src/workflows/Workflow.ts

@@ -1,10 +1,10 @@
 import { IClassOf } from "../decorators/IClassOf.js";
-import Inject from "../di/di.js";
 import DateTime from "../types/DateTime.js";
 import TimeSpan from "../types/TimeSpan.js";
 import { ActivitySuspendedError } from "./ActivitySuspendedError.js";
-import WorkflowContext, { IWorkflowThrottleGroup } from "./WorkflowContext.js";
+import type WorkflowContext from "./WorkflowContext.js";
 import { WorkflowRegistry } from "./WorkflowRegistry.js";
+import type { IWorkflowThrottleGroup } from "./WorkflowStorage.js";
 
 
 export function Activity(target, key) {

src/workflows/WorkflowContext.ts

@@ -12,7 +12,7 @@ import TimeSpan from "../types/TimeSpan.js";
 import sleep from "../common/sleep.js";
 import Waiter from "./Waiter.js";
 import { WorkflowItem } from "./WorkflowDbContext.js";
-import WorkflowStorage from "./WorkflowStorage.js";
+import WorkflowStorage, { IWorkflowThrottleGroup } from "./WorkflowStorage.js";
 
 export const runChildSymbol = Symbol("runChild");
 
@@ -142,11 +142,6 @@ export interface IWorkflowResult<T> {
     queued?: DateTime;
 }
 
-export interface IWorkflowThrottleGroup {
-    group: string;
-    maxPerSecond: number;
-}
-
 export interface IWorkflowQueueParameter {
     id?: string;
     throwIfExists?: boolean;
@@ -232,9 +227,17 @@ export default class WorkflowContext {
         let throttleGroup = null;
 
         if (throttle) {
-            eta = await this.storage.getNextEta(throttle);
-            queued = eta;
-            throttleGroup = throttle.group;
+            const { deferSeconds } = throttle;
+            if (deferSeconds) {
+                const lw = await this.storage.getLastEta(throttle);
+                if (lw) {
+                    return lw.id;
+                }
+            } else {
+                eta = await this.storage.getNextEta(throttle);
+                queued = eta;
+                throttleGroup = throttle.group;
+            }
         }
 
         // this will ensure even empty workflow !!

src/workflows/WorkflowStorage.ts

@@ -10,6 +10,15 @@ import { loadedFromDb, WorkflowDbContext, WorkflowItem } from "./WorkflowDbConte
 import WorkflowTask from "./WorkflowTask.js";
 import Sql from "../sql/Sql.js";
 
+export type IWorkflowThrottleGroup = {
+    group: string;
+    deferSeconds?: number;
+    maxPerSecond?: never;
+} | {
+    group: string;
+    deferSeconds?: never;
+    maxPerSecond?: number;
+};
 
 @RegisterSingleton
 export default class WorkflowStorage {
@@ -36,9 +45,22 @@ export default class WorkflowStorage {
         return q.count();
     }
 
-    async getNextEta(throttle: { group: string, maxPerSecond: number }) {
+    async getLastEta(throttle: IWorkflowThrottleGroup) {
+        const db = new WorkflowDbContext(this.driver);
+        const w = await db.workflows.where(throttle, (p) => (x) => x.throttleGroup === p.group
+                && x.state !== "failed"
+                && x.state !== "done"
+            )
+            .orderByDescending(void 0, (p) => (x) => x.queued)
+            .first();
+        return w;
+
+    }
+
+    async getNextEta(throttle: IWorkflowThrottleGroup) {
 
         const db = new WorkflowDbContext(this.driver);
+
         const last = await db.workflows.where(throttle, (p) => (x) => x.throttleGroup === p.group
             && x.isWorkflow === true)
             .orderByDescending(void 0, (p) => (x) => x.queued)