Add workflow deletion signal (#1880)

Part of OPS-3195

## Additional notes
- Add a cache-based mechanism to detect workflow deletion requests and
stop workflow executions early.
- Add helpers to signal and detect workflow deletion using a cache key
(workflow:deleted:<flowId>)
- This ensures in-progress and future runs don’t continue executing
after a workflow deletion is requested, avoiding wasted work and
inconsistent state.

Author: MarceloRGonc

packages/engine/src/lib/workflow-deletion-validator.ts

@@ -0,0 +1,23 @@
+import { logger, wasWorkflowDeletionRequested } from '@openops/server-shared';
+import { FlowId } from '@openops/shared';
+
+const defaultMessage =
+  'Workflow deletion was requested. Stopping workflow executions.';
+
+export class WorkflowDeletionRequestedError extends Error {
+  constructor(message = defaultMessage) {
+    super(message);
+    this.name = 'WorkflowDeletionRequestedError';
+  }
+}
+
+export async function throwIfWorkflowDeletionRequested(
+  flowId: FlowId,
+): Promise<void> {
+  const wasDeletionRequested = await wasWorkflowDeletionRequested(flowId);
+
+  if (wasDeletionRequested) {
+    logger.info(defaultMessage);
+    throw new WorkflowDeletionRequestedError();
+  }
+}

packages/server/shared/src/index.ts

@@ -31,3 +31,4 @@ export * from './lib/system';
 export * from './lib/truncation-utils';
 export * from './lib/webhook-secrets-util';
 export * from './lib/workflow-cancellation-manager';
+export * from './lib/workflow-deletion-manager';

packages/server/shared/src/lib/workflow-deletion-manager.ts

@@ -0,0 +1,25 @@
+import { FlowId } from '@openops/shared';
+import { cacheWrapper } from './cache/cache-wrapper';
+
+const DELETED_PREFIX = 'workflow:deleted:';
+
+function buildCancellationKey(flowId: FlowId): string {
+  return `${DELETED_PREFIX}${flowId}`;
+}
+
+export async function signalWorkflowDeletion(flowId: FlowId): Promise<void> {
+  const key = buildCancellationKey(flowId);
+  await cacheWrapper.setKey(key, 'true');
+}
+
+export async function wasWorkflowDeletionRequested(
+  flowId: FlowId,
+): Promise<boolean> {
+  const key = buildCancellationKey(flowId);
+  const value: string | null = await cacheWrapper.getKey(key);
+  if (value === null) {
+    return false;
+  }
+
+  return value === 'true';
+}