Update Workflow Delay (#1217)

Closes #1216

Also do a big cleanup of dependencies to fix all Depandabot issues.

Author: kraftp

package-lock.json

@@ -17,7 +17,7 @@
     "test": "tsx --test tests/*.test.ts"
   },
   "dependencies": {
-    "@confluentinc/kafka-javascript": "^1.4.0"
+    "@confluentinc/kafka-javascript": "^1.8.2"
   },
   "peerDependencies": {
     "@dbos-inc/dbos-sdk": "*"

packages/confluent-kafka-receive/package.json

@@ -26,7 +26,6 @@
   "devDependencies": {
     "@types/jest": "^29.5.14",
     "@types/pg": "^8.15.2",
-    "drizzle-kit": "^0.31.5",
     "drizzle-orm": "^0.44.5",
     "jest": "^29.7.0",
     "typescript": "^5.4.5"

packages/drizzle-datasource/package.json

@@ -5,8 +5,6 @@ import { dropDB, ensureDB } from './test-helpers';
 import { randomUUID } from 'crypto';
 import SuperJSON from 'superjson';
 import { pgTable, text, integer } from 'drizzle-orm/pg-core';
-import { drizzle } from 'drizzle-orm/node-postgres';
-import { pushSchema } from 'drizzle-kit/api';
 import { sql } from 'drizzle-orm';
 
 const config = { user: 'postgres', database: 'drizzle_pool_test_userdb' };
@@ -69,14 +67,9 @@ describe('DrizzleDataSource with custom Pool', () => {
       poolClient.release();
     }
 
-    const drizzlePool = new Pool(config);
-    const db = drizzle(drizzlePool);
-    try {
-      const res = await pushSchema({ greetingsTable }, db);
-      await res.apply();
-    } finally {
-      await drizzlePool.end();
-    }
+    await customPool.query(
+      `CREATE TABLE IF NOT EXISTS greetings (name TEXT PRIMARY KEY NOT NULL, greet_count INTEGER DEFAULT 0)`,
+    );
   });
 
   afterAll(async () => {

packages/drizzle-datasource/tests/custom-pool.test.ts

@@ -5,8 +5,6 @@ import { dropDB, ensureDB } from './test-helpers';
 import { randomUUID } from 'crypto';
 import SuperJSON from 'superjson';
 import { pgTable, text, integer } from 'drizzle-orm/pg-core';
-import { drizzle } from 'drizzle-orm/node-postgres';
-import { pushSchema } from 'drizzle-kit/api';
 import { eq, sql } from 'drizzle-orm';
 
 const config = { user: 'postgres', database: 'drizzle_ds_test_userdb' };
@@ -19,14 +17,14 @@ interface transaction_completion {
   error: string | null;
 }
 
-async function createSchema(config: PoolConfig, entities: { [key: string]: object }) {
-  const drizzlePool = new Pool(config);
-  const db = drizzle(drizzlePool);
+async function createSchema(config: PoolConfig) {
+  const pool = new Pool(config);
   try {
-    const res = await pushSchema(entities, db);
-    await res.apply();
+    await pool.query(
+      `CREATE TABLE IF NOT EXISTS greetings (name TEXT PRIMARY KEY NOT NULL, greet_count INTEGER DEFAULT 0)`,
+    );
   } finally {
-    await drizzlePool.end();
+    await pool.end();
   }
 }
 
@@ -344,7 +342,7 @@ async function createDatabases(createTxCompletions: boolean) {
   if (createTxCompletions) {
     await DrizzleDataSource.initializeDBOSSchema(config);
   }
-  await createSchema(config, { greetingsTable });
+  await createSchema(config);
 }
 
 async function insertFunction(user: string) {

packages/drizzle-datasource/tests/datasource.test.ts

@@ -399,6 +399,10 @@ export class DBOSClient {
     return this.systemDatabase.setWorkflowPriority(workflowID, priority);
   }
 
+  setWorkflowDelay(workflowID: string, delaySeconds: number): Promise<void> {
+    return this.systemDatabase.setWorkflowDelay(workflowID, delaySeconds);
+  }
+
   deleteWorkflow(workflowID: string, deleteChildren: boolean = false): Promise<void> {
     return this.systemDatabase.deleteWorkflows([workflowID], deleteChildren);
   }

src/client.ts

@@ -816,6 +816,23 @@ export class DBOS {
     }, 'DBOS.setWorkflowPriority');
   }
 
+  /**
+   * Update the delay on a DELAYED workflow. The new delay is calculated from when this method is called.
+   * The workflow will remain in DELAYED status until the new delay expires, then transition to ENQUEUED.
+   * Only affects workflows with DELAYED status.
+   * @param workflowID - ID of the workflow
+   * @param delaySeconds - Number of seconds to delay the workflow from now.
+   */
+  static async setWorkflowDelay(workflowID: string, delaySeconds: number): Promise<void> {
+    ensureDBOSIsLaunched('setWorkflowDelay');
+    if (delaySeconds < 0) {
+      throw new DBOSError('delaySeconds must be non-negative');
+    }
+    return runInternalStep(async () => {
+      return DBOSExecutor.globalInstance!.systemDatabase.setWorkflowDelay(workflowID, delaySeconds);
+    }, 'DBOS.setWorkflowDelay');
+  }
+
   /**
    * Retrieve the steps of a workflow
    * @param workflowID - ID of the workflow

src/dbos.ts

@@ -994,6 +994,17 @@ export class SystemDatabase {
     );
   }
 
+  async setWorkflowDelay(workflowID: string, delaySeconds: number): Promise<void> {
+    const delayUntilEpochMS = Date.now() + delaySeconds * 1000;
+    await this.pool.query(
+      `UPDATE "${this.schemaName}".workflow_status
+       SET delay_until_epoch_ms = $1, updated_at = $2
+       WHERE workflow_uuid = $3
+         AND status = $4`,
+      [delayUntilEpochMS, Date.now(), workflowID, StatusString.DELAYED],
+    );
+  }
+
   async getWorkflowChildren(workflowID: string): Promise<string[]> {
     // BFS to find all descendant workflows
     const visited = new Set<string>([workflowID]);

src/system_database.ts

@@ -1955,4 +1955,32 @@ describe('delay-tests', () => {
     expect(await resumeHandle.getResult()).toBe('done');
     expect((await resumeHandle.getStatus())?.status).toBe(StatusString.SUCCESS);
   }, 15000);
+
+  test('test_setWorkflowDelay', async () => {
+    // Start a workflow with a long delay
+    const handle = await DBOS.startWorkflow(TestDelayWFs, {
+      queueName: TestDelayWFs.queue.name,
+      enqueueOptions: { delaySeconds: 60 },
+    }).testWorkflowStr();
+    expect((await handle.getStatus())?.status).toBe(StatusString.DELAYED);
+
+    // Update the delay to a short value
+    const tBefore = Date.now();
+    await DBOS.setWorkflowDelay(handle.workflowID, 1);
+    const tAfter = Date.now();
+
+    // Verify the delay was updated
+    const status = await handle.getStatus();
+    expect(status?.status).toBe(StatusString.DELAYED);
+    expect(status?.delayUntilEpochMS).toBeDefined();
+    expect(status!.delayUntilEpochMS!).toBeGreaterThanOrEqual(tBefore + 1000);
+    expect(status!.delayUntilEpochMS!).toBeLessThanOrEqual(tAfter + 1000);
+
+    // Wait for the workflow to complete
+    expect(await handle.getResult()).toBe('done');
+    expect((await handle.getStatus())?.status).toBe(StatusString.SUCCESS);
+
+    // Test invalid delay
+    await expect(DBOS.setWorkflowDelay('some-id', -1)).rejects.toThrow('delaySeconds must be non-negative');
+  }, 30000);
 });