code: private String insertOrUpdateWorkflow(Workflow workflow, boolean update) {
  		Preconditions.checkNotNull(workflow, "workflow object cannot be null");
  
  		if (workflow.getStatus().isTerminal()) {
  			workflow.setEndTime(System.currentTimeMillis());
  		}
  		List<Task> tasks = workflow.getTasks();
  		workflow.setTasks(new LinkedList<>());
  
  		String payload = toJson(workflow);
  		// Store the workflow object
  		dynoClient.set(nsKey(WORKFLOW, workflow.getWorkflowId()), payload);
  		recordRedisDaoRequests("storeWorkflow", "n/a", workflow.getWorkflowName());
  		recordRedisDaoPayloadSize("storeWorkflow", payload.length(), "n/a", workflow.getWorkflowName());
  		if (!update) {
  			// Add to list of workflows for a workflowdef
  			String key = nsKey(WORKFLOW_DEF_TO_WORKFLOWS, workflow.getWorkflowName(), dateStr(workflow.getCreateTime()));
  			dynoClient.sadd(key, workflow.getWorkflowId());
  			if (workflow.getCorrelationId() != null) {
  				// Add to list of workflows for a correlationId
  				dynoClient.sadd(nsKey(CORR_ID_TO_WORKFLOWS, workflow.getCorrelationId()), workflow.getWorkflowId());
  			}
  		}
  		// Add or remove from the pending workflows
  		if (workflow.getStatus().isTerminal()) {
  			dynoClient.srem(nsKey(PENDING_WORKFLOWS, workflow.getWorkflowName()), workflow.getWorkflowId());
  		} else {
  			dynoClient.sadd(nsKey(PENDING_WORKFLOWS, workflow.getWorkflowName()), workflow.getWorkflowId());
  		}
  
  		workflow.setTasks(tasks);
  		return workflow.getWorkflowId();
  	}
docs: Inserts a new workflow/ updates an existing workflow in the datastore.
  Additionally, if a workflow is in terminal state, it is removed from the set of pending workflows.
  
  @param workflow the workflow instance
  @param update flag to identify if update or create operation
  @return the workflowId