Ran a fix for OrgID sometimes missing during Workflow execution

This commit is contained in:
frikky
2022-08-02 14:46:23 +02:00
parent cfa4960210
commit 2e3ae7d9cf
4 changed files with 30 additions and 15 deletions
+1 -1
View File
@@ -24,7 +24,7 @@ require (
github.com/h2non/filetype v1.1.3 github.com/h2non/filetype v1.1.3
github.com/nirasan/go-oauth-pkce-code-verifier v0.0.0-20170819232839-0fbfe93532da // indirect github.com/nirasan/go-oauth-pkce-code-verifier v0.0.0-20170819232839-0fbfe93532da // indirect
github.com/satori/go.uuid v1.2.0 github.com/satori/go.uuid v1.2.0
github.com/shuffle/shuffle-shared v0.2.77 github.com/shuffle/shuffle-shared v0.2.78
go4.org v0.0.0-20201209231011-d4a079459e60 // indirect go4.org v0.0.0-20201209231011-d4a079459e60 // indirect
golang.org/x/crypto v0.0.0-20220112180741-5e0467b6c7ce golang.org/x/crypto v0.0.0-20220112180741-5e0467b6c7ce
google.golang.org/api v0.65.0 google.golang.org/api v0.65.0
+15 -5
View File
@@ -2311,7 +2311,7 @@ func handleWebhookCallback(resp http.ResponseWriter, request *http.Request) {
//start, startok := request.URL.Query()["start"] //start, startok := request.URL.Query()["start"]
// OrgId: activeOrgs[0].Id, // OrgId: activeOrgs[0].Id,
workflowExecution, executionResp, err := handleExecution(item, workflow, newRequest) workflowExecution, executionResp, err := handleExecution(item, workflow, newRequest, hook.OrgId)
if err == nil { if err == nil {
/* /*
err = increaseStatisticsField(ctx, "total_webhooks_ran", workflowExecution.Workflow.ID, 1, workflowExecution.ExecutionOrg) err = increaseStatisticsField(ctx, "total_webhooks_ran", workflowExecution.Workflow.ID, 1, workflowExecution.ExecutionOrg)
@@ -3585,7 +3585,7 @@ func handleCloudExecutionOnprem(workflowId, startNode, executionSource, executio
Body: ioutil.NopCloser(bytes.NewReader(b)), Body: ioutil.NopCloser(bytes.NewReader(b)),
} }
_, _, err = handleExecution(workflowId, shuffle.Workflow{}, newRequest) _, _, err = handleExecution(workflowId, shuffle.Workflow{}, newRequest, workflow.OrgId)
return err return err
} }
@@ -3706,7 +3706,7 @@ func handleCloudJob(job shuffle.CloudSyncJob) error {
return err return err
} }
_, _, err = handleExecution(job.PrimaryItemId, shuffle.Workflow{}, newRequest) _, _, err = handleExecution(job.PrimaryItemId, shuffle.Workflow{}, newRequest, job.OrgId)
if err != nil { if err != nil {
log.Printf("Failed continuing workflow from cloud user_input: %s", err) log.Printf("Failed continuing workflow from cloud user_input: %s", err)
return err return err
@@ -4009,7 +4009,12 @@ func runInitEs(ctx context.Context) {
Body: ioutil.NopCloser(strings.NewReader(schedule.WrappedArgument)), Body: ioutil.NopCloser(strings.NewReader(schedule.WrappedArgument)),
} }
_, _, err := handleExecution(schedule.WorkflowId, shuffle.Workflow{}, request) orgId := ""
if len(activeOrgs) > 0 {
orgId = activeOrgs[0].Id
}
_, _, err := handleExecution(schedule.WorkflowId, shuffle.Workflow{}, request, orgId)
if err != nil { if err != nil {
log.Printf("[WARNING] Failed to execute %s: %s", schedule.WorkflowId, err) log.Printf("[WARNING] Failed to execute %s: %s", schedule.WorkflowId, err)
} }
@@ -4869,7 +4874,12 @@ func runInit(ctx context.Context) {
Body: ioutil.NopCloser(strings.NewReader(schedule.WrappedArgument)), Body: ioutil.NopCloser(strings.NewReader(schedule.WrappedArgument)),
} }
_, _, err := handleExecution(schedule.WorkflowId, shuffle.Workflow{}, request) orgId := ""
if len(activeOrgs) > 0 {
orgId = activeOrgs[0].Id
}
_, _, err := handleExecution(schedule.WorkflowId, shuffle.Workflow{}, request, orgId)
if err != nil { if err != nil {
log.Printf("[WARNING] Failed to execute %s: %s", schedule.WorkflowId, err) log.Printf("[WARNING] Failed to execute %s: %s", schedule.WorkflowId, err)
} }
+9 -5
View File
@@ -98,7 +98,7 @@ func createSchedule(ctx context.Context, scheduleId, workflowId, name, startNode
Body: ioutil.NopCloser(strings.NewReader(bodyWrapper)), Body: ioutil.NopCloser(strings.NewReader(bodyWrapper)),
} }
_, _, err := handleExecution(workflowId, shuffle.Workflow{ExecutingOrg: shuffle.OrgMini{Id: orgId}}, request) _, _, err := handleExecution(workflowId, shuffle.Workflow{ExecutingOrg: shuffle.OrgMini{Id: orgId}}, request, orgId)
if err != nil { if err != nil {
log.Printf("Failed to execute %s: %s", workflowId, err) log.Printf("Failed to execute %s: %s", workflowId, err)
} }
@@ -831,7 +831,7 @@ func getWorkflowLocal(fileId string, request *http.Request) ([]byte, error) {
return body, nil return body, nil
} }
func handleExecution(id string, workflow shuffle.Workflow, request *http.Request) (shuffle.WorkflowExecution, string, error) { func handleExecution(id string, workflow shuffle.Workflow, request *http.Request, orgId string) (shuffle.WorkflowExecution, string, error) {
//go func() { //go func() {
// log.Printf("\n\nPRE TIME: %s\n\n", time.Now().Format("2006-01-02 15:04:05")) // log.Printf("\n\nPRE TIME: %s\n\n", time.Now().Format("2006-01-02 15:04:05"))
// _ = <-time.After(time.Second * 60) // _ = <-time.After(time.Second * 60)
@@ -850,8 +850,12 @@ func handleExecution(id string, workflow shuffle.Workflow, request *http.Request
} }
if len(workflow.ExecutingOrg.Id) == 0 { if len(workflow.ExecutingOrg.Id) == 0 {
log.Printf("[INFO] Stopped execution because there is no executing org for workflow %s", workflow.ID) if len(orgId) > 0 {
return shuffle.WorkflowExecution{}, fmt.Sprintf("Workflow has no executing org defined"), errors.New("Workflow has no executing org defined") workflow.ExecutingOrg.Id = orgId
} else {
log.Printf("[INFO] Stopped execution because there is no executing org for workflow %s", workflow.ID)
return shuffle.WorkflowExecution{}, fmt.Sprintf("Workflow has no executing org defined"), errors.New("Workflow has no executing org defined")
}
} }
if len(workflow.Actions) == 0 { if len(workflow.Actions) == 0 {
@@ -1138,7 +1142,7 @@ func executeWorkflow(resp http.ResponseWriter, request *http.Request) {
user.ActiveOrg.Users = []shuffle.UserMini{} user.ActiveOrg.Users = []shuffle.UserMini{}
workflow.ExecutingOrg = user.ActiveOrg workflow.ExecutingOrg = user.ActiveOrg
workflowExecution, executionResp, err := handleExecution(fileId, *workflow, request) workflowExecution, executionResp, err := handleExecution(fileId, *workflow, request, user.ActiveOrg.Id)
if err == nil { if err == nil {
resp.WriteHeader(200) resp.WriteHeader(200)
resp.Write([]byte(fmt.Sprintf(`{"success": true, "execution_id": "%s", "authorization": "%s"}`, workflowExecution.ExecutionId, workflowExecution.Authorization))) resp.Write([]byte(fmt.Sprintf(`{"success": true, "execution_id": "%s", "authorization": "%s"}`, workflowExecution.ExecutionId, workflowExecution.Authorization)))
+5 -4
View File
@@ -2641,11 +2641,12 @@ const AngularWorkflow = (defaultprops) => {
setSelectedApp(tmpapp); setSelectedApp(tmpapp);
setSelectedAction(curaction); setSelectedAction(curaction);
} else { } else {
//if (curapp.id !== curaction.id) {
// curaction.app_id = curapp.id
// //.valueOf()
//}
console.log("CURAPP: ", curapp) console.log("CURAPP: ", curapp)
if (curapp.id !== curaction.id) { curaction.app_id = curapp.id
curaction.app_id = curapp.id
//.valueOf()
}
setAuthenticationType( setAuthenticationType(
curapp.authentication.type === "oauth2" && curapp.authentication.redirect_uri !== undefined && curapp.authentication.redirect_uri !== null ? { curapp.authentication.type === "oauth2" && curapp.authentication.redirect_uri !== undefined && curapp.authentication.redirect_uri !== null ? {