Change executed task query

src/external-api/conductor-network-types.ts

@@ -118,6 +118,7 @@ const ExecutedWorkflowTask = t.type({
   taskType: optional(t.string),
   status: optional(ExecutedWorkflowTaskStatus),
   inputData: optional(t.record(t.string, t.unknown)),
+  outputData: optional(t.record(t.string, t.unknown)),
   referenceTaskName: optional(t.string),
   retryCount: optional(t.number),
   seq: optional(t.number),
@@ -163,7 +164,7 @@ const ExecutedWorkflow = t.type({
   updatedBy: optional(t.string),
   status: ExecutedWorkflowStatus,
   endTime: optional(t.number),
-  workflowId: optional(t.string),
+  workflowId: t.string,
   parentWorkflowId: optional(t.string),
   parentWorkflowTaskId: optional(t.string),
   tasks: optional(t.array(ExecutedWorkflowTask)),
@@ -264,3 +265,7 @@ export function decodeBulkRetryOutput(value: unknown): BulkOperationOutput {
 export function decodeBulkRestartOutput(value: unknown): BulkOperationOutput {
   return extractResult(BulkOperation.decode(value));
 }
+
+export function decodeExecutedWorkflowTaskDetailOutput(value: unknown): ApiExecutedWorkflowTask {
+  return extractResult(ExecutedWorkflowTask.decode(value));
+}

src/external-api/conductor.ts

@@ -15,6 +15,8 @@ import {
   TaskDefinitionsOutput,
   decodeTaskDefinitionsOutput,
   decodeBulkTerminateOutput,
+  decodeExecutedWorkflowTaskDetailOutput,
+  ApiExecutedWorkflowTask,
 } from './conductor-network-types';
 import { sendDeleteRequest, sendGetRequest, sendPostRequest, sendPutRequest } from './helpers';
 
@@ -180,7 +182,12 @@ async function executeWorkflowByName(
   { name, inputParameters, correlationId, version, priority }: ExecuteWorkflowByNameInput,
 ): Promise<string> {
   const executedWorkflowId = await sendPostRequest(
-    [baseURL, `workflow/${name}?version=${version}&correlationId=${correlationId}&priority=${priority}`],
+    [
+      baseURL,
+      `workflow/${name}?${version == null ? '' : `version=${version}`}${
+        correlationId == null ? '' : `&correlationId=${correlationId}`
+      }&priority=${priority == null ? 0 : priority}`,
+    ],
     inputParameters,
   );
 
@@ -197,6 +204,12 @@ async function getTaskDefinitions(baseURL: string): Promise<TaskDefinitionsOutpu
   return data;
 }
 
+async function getExecutedWorkflowTaskDetail(baseURL: string, taskId: string): Promise<ApiExecutedWorkflowTask> {
+  const json = await sendGetRequest([baseURL, `/tasks/${taskId}`]);
+  const data = decodeExecutedWorkflowTaskDetailOutput(json);
+  return data;
+}
+
 const conductorAPI = {
   getWorkflowMetadata,
   getWorkflowDetail,
@@ -219,6 +232,7 @@ const conductorAPI = {
   executeNewWorkflow,
   executeWorkflowByName,
   getTaskDefinitions,
+  getExecutedWorkflowTaskDetail,
 };
 
 export type ConductorAPI = typeof conductorAPI;

src/external-api/helpers.ts

@@ -89,10 +89,13 @@ async function apiFetch(path: APIPath, options: RequestInit): Promise<unknown> {
     return response;
   }
 
-  const json = JSON.parse(text);
-  logResponse(requestId, json);
-
-  return json;
+  try {
+    const json = JSON.parse(text);
+    return json;
+  } catch (e) {
+    logResponse(requestId, text);
+    return text;
+  }
 }
 
 export async function sendGetRequest(path: APIPath, cookie?: string): Promise<unknown> {

src/helpers/workflow.helpers.ts

@@ -112,7 +112,7 @@ function extractSubworkflowsFromTasks(task: ExecutedWorkflowTask): SubWorkflow |
 }
 
 type SubworkflowDetail = {
-  taskReferenceName: string;
+  referenceTaskName: string;
   workflowDetail: Workflow;
   executedWorkflowDetail: ExecutedWorkflow;
 };
@@ -137,7 +137,7 @@ async function getSubworklowsDetail(subWorkflow: SubWorkflow): Promise<Subworkfl
   };
 
   return {
-    taskReferenceName: referenceTaskName,
+    referenceTaskName,
     workflowDetail: workflowDetailWithId,
     executedWorkflowDetail: executedWorkflowDetailWithId,
   };

src/schema/api.graphql

@@ -334,6 +334,7 @@ input ExecuteWorkflowByName {
 }
 
 type ExecutedWorkflow implements Node {
+  correlationId: String
   createdAt: String
   createdBy: String
   endTime: String
@@ -353,7 +354,7 @@ type ExecutedWorkflow implements Node {
   variables: String
   version: Int
   workflowDefinition: Workflow
-  workflowId: String
+  workflowId: String!
   workflowName: String
   workflowVersion: Int
 }
@@ -396,20 +397,27 @@ enum ExecutedWorkflowStatus {
 }
 
 type ExecutedWorkflowTask implements Node {
+  callbackAfterSeconds: Int
   endTime: String
   executed: Boolean
+  externalInputPayloadStoragePath: String
+  externalOutputPayloadStoragePath: String
   id: ID!
+  inputData: String
+  outputData: String
+  pollCount: Int
   reasonForIncompletion: String
+  referenceTaskName: String
   retried: Boolean
   retryCount: Int
   scheduledTime: String
+  seq: Int
   startTime: String
   status: ExecutedWorkflowTaskStatus
   subWorkflowId: String
   taskDefName: String
   taskDefinition: String
   taskId: String
-  taskReferenceName: String
   taskType: String
   updateTime: String
   version: Int
@@ -768,7 +776,7 @@ input StartWorkflowRequestInput {
 
 type SubWorkflow {
   executedWorkflowDetail: ExecutedWorkflow!
-  taskReferenceName: String!
+  referenceTaskName: String!
   workflowDetail: Workflow!
 }
 

src/schema/global-types.ts

@@ -130,6 +130,15 @@ export const NodeQuery = extendType({
 
             return { ...schedule, id: args.id, __typename: 'Schedule' };
           }
+          case 'ExecutedWorkflowTask': {
+            const id = fromGraphId('ExecutedWorkflowTask', args.id);
+            const task = await conductorAPI.getExecutedWorkflowTaskDetail(config.conductorApiURL, id);
+            if (task == null) {
+              return null;
+            }
+
+            return { ...task, id: args.id, __typename: 'ExecutedWorkflowTask' };
+          }
           /* eslint-enable */
           default:
             return null;

src/schema/nexus-typegen.ts

@@ -661,7 +661,7 @@ export interface NexusGenObjects {
   SubWorkflow: {
     // root type
     executedWorkflowDetail: NexusGenRootTypes['ExecutedWorkflow']; // ExecutedWorkflow!
-    taskReferenceName: string; // String!
+    referenceTaskName: string; // String!
     workflowDetail: NexusGenRootTypes['Workflow']; // Workflow!
   };
   Subscription: {};
@@ -966,6 +966,7 @@ export interface NexusGenFieldTypes {
   };
   ExecutedWorkflow: {
     // field return type
+    correlationId: string | null; // String
     createdAt: string | null; // String
     createdBy: string | null; // String
     endTime: string | null; // String
@@ -985,7 +986,7 @@ export interface NexusGenFieldTypes {
     variables: string | null; // String
     version: number | null; // Int
     workflowDefinition: NexusGenRootTypes['Workflow'] | null; // Workflow
-    workflowId: string | null; // String
+    workflowId: string; // String!
     workflowName: string | null; // String
     workflowVersion: number | null; // Int
   };
@@ -1002,20 +1003,27 @@ export interface NexusGenFieldTypes {
   };
   ExecutedWorkflowTask: {
     // field return type
+    callbackAfterSeconds: number | null; // Int
     endTime: string | null; // String
     executed: boolean | null; // Boolean
+    externalInputPayloadStoragePath: string | null; // String
+    externalOutputPayloadStoragePath: string | null; // String
     id: string; // ID!
+    inputData: string | null; // String
+    outputData: string | null; // String
+    pollCount: number | null; // Int
     reasonForIncompletion: string | null; // String
+    referenceTaskName: string | null; // String
     retried: boolean | null; // Boolean
     retryCount: number | null; // Int
     scheduledTime: string | null; // String
+    seq: number | null; // Int
     startTime: string | null; // String
     status: NexusGenEnums['ExecutedWorkflowTaskStatus'] | null; // ExecutedWorkflowTaskStatus
     subWorkflowId: string | null; // String
     taskDefName: string | null; // String
     taskDefinition: string | null; // String
     taskId: string | null; // String
-    taskReferenceName: string | null; // String
     taskType: string | null; // String
     updateTime: string | null; // String
     version: number | null; // Int
@@ -1258,7 +1266,7 @@ export interface NexusGenFieldTypes {
   SubWorkflow: {
     // field return type
     executedWorkflowDetail: NexusGenRootTypes['ExecutedWorkflow']; // ExecutedWorkflow!
-    taskReferenceName: string; // String!
+    referenceTaskName: string; // String!
     workflowDetail: NexusGenRootTypes['Workflow']; // Workflow!
   };
   Subscription: {
@@ -1605,6 +1613,7 @@ export interface NexusGenFieldTypeNames {
   };
   ExecutedWorkflow: {
     // field return type name
+    correlationId: 'String';
     createdAt: 'String';
     createdBy: 'String';
     endTime: 'String';
@@ -1641,20 +1650,27 @@ export interface NexusGenFieldTypeNames {
   };
   ExecutedWorkflowTask: {
     // field return type name
+    callbackAfterSeconds: 'Int';
     endTime: 'String';
     executed: 'Boolean';
+    externalInputPayloadStoragePath: 'String';
+    externalOutputPayloadStoragePath: 'String';
     id: 'ID';
+    inputData: 'String';
+    outputData: 'String';
+    pollCount: 'Int';
     reasonForIncompletion: 'String';
+    referenceTaskName: 'String';
     retried: 'Boolean';
     retryCount: 'Int';
     scheduledTime: 'String';
+    seq: 'Int';
     startTime: 'String';
     status: 'ExecutedWorkflowTaskStatus';
     subWorkflowId: 'String';
     taskDefName: 'String';
     taskDefinition: 'String';
     taskId: 'String';
-    taskReferenceName: 'String';
     taskType: 'String';
     updateTime: 'String';
     version: 'Int';
@@ -1897,7 +1913,7 @@ export interface NexusGenFieldTypeNames {
   SubWorkflow: {
     // field return type name
     executedWorkflowDetail: 'ExecutedWorkflow';
-    taskReferenceName: 'String';
+    referenceTaskName: 'String';
     workflowDetail: 'Workflow';
   };
   Subscription: {

src/schema/task.ts

@@ -53,7 +53,7 @@ export const ExecutedWorkflowTask = objectType({
       resolve: (executedWorkflowTask) => toGraphId('ExecutedWorkflowTask', executedWorkflowTask.taskId ?? uuid()),
     });
     t.string('taskType');
-    t.string('taskReferenceName');
+    t.string('referenceTaskName');
     t.field('status', { type: ExecutedWorkflowTaskStatus });
     t.int('retryCount');
     t.string('startTime', {
@@ -76,6 +76,13 @@ export const ExecutedWorkflowTask = objectType({
     t.string('reasonForIncompletion');
     t.string('taskDefinition', { resolve: (task) => JSON.stringify(task.taskDefinition) });
     t.string('subWorkflowId');
+    t.string('inputData', { resolve: (task) => JSON.stringify(task.inputData) });
+    t.string('outputData', { resolve: (task) => JSON.stringify(task.outputData) });
+    t.string('externalOutputPayloadStoragePath');
+    t.string('externalInputPayloadStoragePath');
+    t.int('callbackAfterSeconds');
+    t.int('seq');
+    t.int('pollCount');
   },
 });