filter

Author: kraftp

src/conductor/conductor.ts

@@ -773,6 +773,15 @@ export class Conductor {
                 executorId: aggBody.executor_id,
                 queueName: aggBody.queue_name,
                 workflowIdPrefix: aggBody.workflow_id_prefix,
+                workflowIDs: aggBody.workflow_uuids,
+                authenticatedUser: aggBody.authenticated_user,
+                forkedFrom: aggBody.forked_from,
+                parentWorkflowID: aggBody.parent_workflow_id,
+                scheduleName: aggBody.schedule_name,
+                queuesOnly: aggBody.queues_only,
+                wasForkedFrom: aggBody.was_forked_from,
+                hasParent: aggBody.has_parent,
+                attributes: aggBody.attributes,
               });
               aggOutput = rows.map((r) => ({
                 group: r.group,

src/conductor/protocol.ts

@@ -862,6 +862,15 @@ export interface GetWorkflowAggregatesBody {
   executor_id?: string[];
   queue_name?: string[];
   workflow_id_prefix?: string[];
+  workflow_uuids?: string[];
+  authenticated_user?: string[];
+  forked_from?: string[];
+  parent_workflow_id?: string[];
+  schedule_name?: string[];
+  queues_only?: boolean;
+  was_forked_from?: boolean;
+  has_parent?: boolean;
+  attributes?: Record<string, unknown>;
 }
 
 export class GetWorkflowAggregatesRequest implements BaseMessage {

src/system_database.ts

@@ -167,6 +167,15 @@ export interface GetWorkflowAggregatesInput {
   executorId?: string[];
   queueName?: string[];
   workflowIdPrefix?: string[];
+  workflowIDs?: string[];
+  authenticatedUser?: string[];
+  forkedFrom?: string[];
+  wasForkedFrom?: boolean;
+  parentWorkflowID?: string[];
+  hasParent?: boolean;
+  queuesOnly?: boolean;
+  attributes?: Record<string, unknown>;
+  scheduleName?: string[];
 }
 
 export interface GetStepAggregatesInput {
@@ -2982,6 +2991,37 @@ export class SystemDatabase {
       whereClauses.push(`(${likeClauses.join(' OR ')})`);
     }
 
+    addFilter('workflow_uuid', input.workflowIDs);
+    addFilter('authenticated_user', input.authenticatedUser);
+    addFilter('forked_from', input.forkedFrom);
+    addFilter('parent_workflow_id', input.parentWorkflowID);
+    addFilter('schedule_name', input.scheduleName);
+
+    // Only workflows that are actively enqueued.
+    if (input.queuesOnly) {
+      whereClauses.push(`queue_name IS NOT NULL`);
+      whereClauses.push(`status IN ($${paramIdx}, $${paramIdx + 1}, $${paramIdx + 2})`);
+      params.push(StatusString.ENQUEUED, StatusString.PENDING, StatusString.DELAYED);
+      paramIdx += 3;
+    }
+
+    if (input.wasForkedFrom !== undefined) {
+      whereClauses.push(`was_forked_from = $${paramIdx}`);
+      params.push(input.wasForkedFrom);
+      paramIdx++;
+    }
+
+    if (input.hasParent !== undefined) {
+      whereClauses.push(input.hasParent ? `parent_workflow_id IS NOT NULL` : `parent_workflow_id IS NULL`);
+    }
+
+    // Match workflows whose attributes JSONB contains all the given key-value pairs.
+    if (input.attributes && Object.keys(input.attributes).length > 0) {
+      whereClauses.push(`attributes @> $${paramIdx}::jsonb`);
+      params.push(JSON.stringify(input.attributes));
+      paramIdx++;
+    }
+
     if (input.startTime) {
       whereClauses.push(`created_at >= $${paramIdx}`);
       params.push(new Date(input.startTime).getTime());