Onboard get workflow API (#133)

Signed-off-by: Tyler Ohlsen <ohltyler@amazon.com>
(cherry picked from commit 28ce545)

Author: ohltyler

common/interfaces.ts

@@ -16,10 +16,6 @@ export type Index = {
  */
 
 export type ReactFlowComponent = Node<IComponentData>;
-
-// TODO: we may not need this re-defined type here at all, if we don't add
-// any special fields/configuration for an edge. Currently this
-// is the same as the default Edge type.
 export type ReactFlowEdge = Edge<{}> & {};
 
 type ReactFlowViewport = {

public/component_types/indexer/indexer.ts

@@ -22,8 +22,6 @@ export class Indexer extends BaseComponent {
       {
         id: 'transformer',
         label: 'Transformer',
-        // TODO: may need to change to be looser. it should be able to take
-        // in other component types
         baseClass: COMPONENT_CLASS.TRANSFORMER,
         acceptMultiple: false,
       },

public/component_types/interfaces.ts

@@ -12,8 +12,6 @@ import { COMPONENT_CATEGORY, COMPONENT_CLASS } from '../utils';
  */
 export type FieldType = 'string' | 'json' | 'select';
 export type SelectType = 'model';
-// TODO: this may expand to more types in the future. Formik supports 'any' so we can too.
-// For now, limiting scope to expected types.
 export type FieldValue = string | {};
 export type ComponentFormValues = FormikValues;
 export type WorkspaceFormValues = {

public/pages/workflow_detail/workflow_detail.tsx

@@ -14,8 +14,8 @@ import { getCore } from '../../services';
 import { WorkflowDetailHeader } from './components';
 import {
   AppState,
+  getWorkflow,
   searchModels,
-  searchWorkflows,
   useAppDispatch,
 } from '../../store';
 import { ResizableWorkspace } from './workspace';
@@ -109,8 +109,7 @@ export function WorkflowDetail(props: WorkflowDetailProps) {
   // - fetch available models as their IDs may be used when building flows
   useEffect(() => {
     if (!isNewWorkflow) {
-      // TODO: can optimize to only fetch a single workflow
-      dispatch(searchWorkflows(FETCH_ALL_QUERY_BODY));
+      dispatch(getWorkflow(workflowId));
     }
     dispatch(searchModels(FETCH_ALL_QUERY_BODY));
   }, []);

public/pages/workflow_detail/workspace/resizable_workspace.tsx

@@ -32,6 +32,7 @@ import {
   WORKFLOW_STATE,
   processNodes,
   reduceToTemplate,
+  ReactFlowEdge,
 } from '../../../../common';
 import { validateWorkspaceFlow, toTemplateFlows } from '../utils';
 import {
@@ -133,8 +134,13 @@ export function ResizableWorkspace(props: ResizableWorkspaceProps) {
    * - open the panel if a node is selected and the panel is closed
    * - it is assumed that only one node can be selected at once
    */
-  // TODO: make more typesafe
-  function onSelectionChange({ nodes, edges }) {
+  function onSelectionChange({
+    nodes,
+    edges,
+  }: {
+    nodes: ReactFlowComponent[];
+    edges: ReactFlowEdge[];
+  }) {
     if (nodes && nodes.length > 0) {
       setSelectedComponent(nodes[0]);
       if (!isDetailsPanelOpen) {
@@ -276,7 +282,6 @@ export function ResizableWorkspace(props: ResizableWorkspaceProps) {
           } as Workflow;
           processWorkflowFn(updatedWorkflow);
         } else {
-          // TODO: bubble up flow error?
           setFlowValidOnSubmit(false);
           setIsSaving(false);
         }
@@ -336,7 +341,10 @@ export function ResizableWorkspace(props: ResizableWorkspaceProps) {
                         setIsDeprovisioning(false);
                       });
                   } else {
-                    // TODO: this case should not happen
+                    // This case should not happen
+                    console.debug(
+                      'Deprovisioning triggered on an invalid workflow. Ignoring.'
+                    );
                   }
                 }}
               >
@@ -360,7 +368,10 @@ export function ResizableWorkspace(props: ResizableWorkspaceProps) {
                         setIsProvisioning(false);
                       });
                   } else {
-                    // TODO: this case should not happen
+                    // This case should not happen
+                    console.debug(
+                      'Provisioning triggered on an invalid workflow. Ignoring.'
+                    );
                   }
                 }}
               >
@@ -370,7 +381,6 @@ export function ResizableWorkspace(props: ResizableWorkspaceProps) {
                 fill={false}
                 disabled={!isSaveable || isLoadingGlobal || isDeprovisionable}
                 isLoading={isSaving}
-                // TODO: if props.isNewWorkflow is true, clear the workflow cache if saving is successful.
                 onClick={() => {
                   setIsSaving(true);
                   dispatch(removeDirty());

public/pages/workflow_detail/workspace/workspace.tsx

@@ -21,6 +21,7 @@ import { setDirty, useAppDispatch } from '../../../store';
 import {
   IComponentData,
   ReactFlowComponent,
+  ReactFlowEdge,
   Workflow,
 } from '../../../../common';
 import {
@@ -41,8 +42,13 @@ interface WorkspaceProps {
   readonly: boolean;
   onNodesChange: (nodes: ReactFlowComponent[]) => void;
   id: string;
-  // TODO: make more typesafe
-  onSelectionChange: ({ nodes, edges }) => void;
+  onSelectionChange: ({
+    nodes,
+    edges,
+  }: {
+    nodes: ReactFlowComponent[];
+    edges: ReactFlowEdge[];
+  }) => void;
 }
 
 const nodeTypes = {

public/store/reducers/workflows_reducer.ts

@@ -220,12 +220,11 @@ const workflowsSlice = createSlice({
       // Fulfilled states: mutate state depending on the action type
       // and payloads
       .addCase(getWorkflow.fulfilled, (state, action) => {
-        // TODO: add logic to mutate state
-        // const workflow = action.payload;
-        // state.workflows = {
-        //   ...state.workflows,
-        //   [workflow.id]: workflow,
-        // };
+        const { workflow } = action.payload;
+        state.workflows = {
+          ...state.workflows,
+          [workflow.id]: workflow,
+        };
         state.loading = false;
         state.errorMessage = '';
       })

public/utils/utils.ts

@@ -154,8 +154,6 @@ export function getComponentSchema(data: IComponentData): ObjectSchema<any> {
   return yup.object(schemaObj);
 }
 
-// TODO: finalize validations for different field types. May need
-// to refer to some backend implementations or OpenSearch documentation
 function getFieldSchema(field: IComponentField): Schema {
   let baseSchema: Schema;
   switch (field.type) {

server/routes/flow_framework_routes_service.ts

@@ -33,6 +33,7 @@ import {
   getWorkflowStateFromResponse,
   getWorkflowsFromResponses,
   isIgnorableError,
+  toWorkflowObj,
 } from './helpers';
 
 /**
@@ -152,7 +153,9 @@ export class FlowFrameworkRoutesService {
     this.client = client;
   }
 
-  // TODO: test e2e
+  // TODO: can remove or simplify if we can fetch all data from a single API call. Tracking issue:
+  // https://github.com/opensearch-project/flow-framework/issues/171
+  // Current implementation is making two calls and combining results via helper fn
   getWorkflow = async (
     context: RequestHandlerContext,
     req: OpenSearchDashboardsRequest,
@@ -163,9 +166,19 @@ export class FlowFrameworkRoutesService {
       const response = await this.client
         .asScoped(req)
         .callAsCurrentUser('flowFramework.getWorkflow', { workflow_id });
-      console.log('response from get workflow: ', response);
-      // TODO: format response
-      return res.ok({ body: response });
+      const workflow = toWorkflowObj(response, workflow_id);
+
+      const stateResponse = await this.client
+        .asScoped(req)
+        .callAsCurrentUser('flowFramework.getWorkflowState', { workflow_id });
+      const state = getWorkflowStateFromResponse(
+        stateResponse.state as typeof WORKFLOW_STATE
+      );
+      const workflowWithState = {
+        ...workflow,
+        state,
+      };
+      return res.ok({ body: { workflow: workflowWithState } });
     } catch (err: any) {
       return generateCustomError(res, err);
     }

server/routes/helpers.ts

@@ -32,12 +32,10 @@ export function isIgnorableError(error: any): boolean {
   return error.body?.error?.type === INDEX_NOT_FOUND_EXCEPTION;
 }
 
-function toWorkflowObj(workflowHit: any): Workflow {
-  // TODO: update schema parsing after hit schema has been updated.
-  // https://github.com/opensearch-project/flow-framework/issues/546
-  const hitSource = workflowHit.fields.filter[0];
+// Convert backend workflow into frontend workflow obj
+export function toWorkflowObj(hitSource: any, id: string): Workflow {
   return {
-    id: workflowHit._id,
+    id,
     name: hitSource.name,
     use_case: hitSource.use_case,
     description: hitSource.description || '',
@@ -59,7 +57,10 @@ export function getWorkflowsFromResponses(
 ): WorkflowDict {
   const workflowDict = {} as WorkflowDict;
   workflowHits.forEach((workflowHit: any) => {
-    workflowDict[workflowHit._id] = toWorkflowObj(workflowHit);
+    // TODO: update schema parsing after hit schema has been updated.
+    // https://github.com/opensearch-project/flow-framework/issues/546
+    const hitSource = workflowHit.fields.filter[0];
+    workflowDict[workflowHit._id] = toWorkflowObj(hitSource, workflowHit._id);
     const workflowStateHit = workflowStateHits.find(
       (workflowStateHit) => workflowStateHit._id === workflowHit._id
     );