Event Correlations Doesn't Work

[jaliyaudagedara]

Consider the following Webhook Definition.

{
    "id": "eventwebhook",
    "name": "Event Webhook",
    "description": "Event Webhook Workflow",
    "version": "1.0.0",
    "specVersion": "0.8",
    "events": [
      {
        "name": "WorkflowStartEvent",
        "source": "https://demo.synpase.com/workflow-start-event",
        "type": "com.synapse.demo/workflow-start-event",
        "correlation": [
          {
            "contextAttributeName": "customerid"
          }
        ]
      },
      {
        "name": "WorkflowResumeEvent",
        "source": "https://demo.synpase.com/workflow-resume-event",
        "type": "com.synapse.demo/workflow-resume-event",
        "correlation": [
          {
            "contextAttributeName": "customerid"
          }
        ]
      }
    ],
    "functions": [
      {
        "name": "createWorkflowInstanceFunction",
        "operation": "https://sswv1z5n-44386.aue.devtunnels.ms/swagger/v1/swagger.json#CreateWorkflowInstance",
        "type": "rest"
      },
      {
        "name": "creditCheckFunction",
        "operation": "https://sswv1z5n-44386.aue.devtunnels.ms/swagger/v1/swagger.json#DoCreditCheck",
        "type": "rest"
      }
    ],
    "start": "StartWorkflow",
    "states": [
      {
        "name": "StartWorkflow",
        "type": "event",
        "onEvents": [
          {
            "actions": [
              {
                "name": "CallCreateWorkflowInstanceFunction",
                "functionRef": {
                  "refName": "createWorkflowInstanceFunction",
                  "arguments": {
                    "body": "${ .customer }"
                  }
                }
              }
            ],
            "eventRefs": [
              "WorkflowStartEvent"
            ]
          }
        ],
        "transition": "CheckCredit"
      },
      {
        "name": "CheckCredit",
        "type": "event",
        "onEvents": [
          {
            "actions": [
              {
                "name": "CallCreditCheckFunction",
                "functionRef": {
                  "refName": "creditCheckFunction",
                  "arguments": {
                    "body": "${ .customer }"
                  }
                }
              }
            ],
            "eventRefs": [
              "WorkflowResumeEvent"
            ]
          }
        ],
        "transition": "EvaluateDecision"
      },
      {
        "name": "EvaluateDecision",
        "type": "switch",
        "dataConditions": [
          {
            "name": "Approved",
            "transition": "StartApplication",
            "condition": "${ .customer | .decision == \"Approved\" }"
          },
          {
            "name": "Denied",
            "transition": "RejectApplication",
            "condition": "${ .customer | .decision == \"Denied\" }"
          }
        ],
        "defaultCondition": {
          "transition": "RejectApplication"
        }
      },
      {
        "name": "StartApplication",
        "type": "sleep",
        "duration": "PT2S",
        "end": true
      },
      {
        "name": "RejectApplication",
        "type": "sleep",
        "duration": "PT2S",
        "end": true
      }
    ]
  }

workflow-start-event

{
    "id": "1b080838-2976-493d-897f-07803944f4d3",
    "specversion": "1.0",
    "source": "https://demo.synpase.com/workflow-start-event",
    "type": "com.synapse.demo/workflow-start-event",
    "datacontenttype": "application/json",
    "customerid": "CUSTOMER-1",
    "data": {
        "customer": {
            "id": "customer1",
            "name": "John Doe",
            "SSN": 123456,
            "yearlyIncome": 50000,
            "address": "123 MyLane, MyCity, MyCountry",
            "employer": "MyCompany"
        }
    }
}

workflow-resume-event

{
    "id": "9a080838-4976-493d-897f-07803944f1d7",
    "specversion": "1.0",
    "source": "https://demo.synpase.com/workflow-resume-event",
    "type": "com.synapse.demo/workflow-resume-event",
    "datacontenttype": "application/json",
    "customerid": "CUSTOMER-1",
    "data": {
        "customer": {
            "id": "customer1",
            "name": "John Doe",
            "SSN": 123456,
            "yearlyIncome": 50000,
            "address": "123 MyLane, MyCity, MyCountry",
            "employer": "MyCompany",
            "decision": "Approved"
        }
    }
}

Intention of the Workflow

• Start workflow from an CloudEvent and create a correlation with customerid and Wait
• Resume workflow from an CloudEvent and create a correlation with customerid

Expected Behavior

• Workflow is started from an CloudEvent with customerid and waits for the next CloudEvent with the same customerid

Actual Behavior

1. Workflow is started from an CloudEvent with customerid and control flow isn't going to the next state
2. Once 1 is fixed, the workflow is resumed from an CloudEvent with customerid, but the correlation doesn't work
3. Once 2 is fixed, the control throws an exception on this line: https://github.com/serverlessworkflow/synapse/blob/cba235fcfe665c6ec9712c5ce592e829870fba38/src/core/Synapse.Application/Commands/Correlations/v1/V1CorrelateEventCommand.cs#L111

Troubleshooting

1. Workflow is started from an CloudEvent with customerid and control flow isn't going to the next state

When the CloudEvent is received to the worker, in it's attributes customerid is populated with the correct value.

But when the V1Event is created from the CloudEvent, customerid is getting set to value of the type instead of the actual value

2. Once 1 is fixed, the workflow is resumed from a CloudEvent with customerid, but the correlation doesn't work

When the worker is trying to correlate, Synapse.Integration.Commands.WorkflowInstances.V1TryCorrelateWorkflowInstanceCommand is getting mapped to Synapse.Application.Commands.WorkflowInstances.V1TryCorrelateWorkflowInstanceCommand. Here again, similar issue as above.

Synapse.Integration.Commands.WorkflowInstances.V1TryCorrelateWorkflowInstanceCommand

Synapse.Application.Commands.WorkflowInstances.V1TryCorrelateWorkflowInstanceCommand

3. Once 2 is fixed, say there are multiple workflow instances started and we want to resume only a particular one by providing customerid, it's finding all the pending workflow instances and one of them satisfies our condition, but since we are foreaching if first one is not the one we want, it throws the exception.

Saved correlations,

[
    {
        "id": "710f4888-858b-428c-b296-5b457718d32c",
        "createdAt": "2023-03-14T18:46:22.5255362",
        "lastModified": "2023-03-14T18:46:22.5255362",
        "activationType": "implicit",
        "conditions": [
            {
                "filters": [
                    {
                        "attributes": {
                            "source": "https://demo.synpase.com/workflow-resume-event",
                            "type": "com.synapse.demo/workflow-resume-event"
                        },
                        "correlationMappings": {
                            "customerid": null
                        }
                    }
                ]
            }
        ],
        "outcome": {
            "type": "correlate",
            "target": "eventwebhook-zxjabedrzkc18plx1wa"
        },
        "contexts": [
            {
                "id": "3352b14a-d913-47dc-a493-f8fc5ae60ac8",
                "mappings": {
                    "customerid": "CUSTOMER-3"
                },
                "pendingEvents": []
            }
        ]
    },
    {
        "id": "aa99de55-adb0-43ef-b456-f15117b8b7ac",
        "createdAt": "2023-03-14T18:46:28.5348247",
        "lastModified": "2023-03-14T18:46:28.5348247",
        "activationType": "implicit",
        "conditions": [
            {
                "filters": [
                    {
                        "attributes": {
                            "source": "https://demo.synpase.com/workflow-resume-event",
                            "type": "com.synapse.demo/workflow-resume-event"
                        },
                        "correlationMappings": {
                            "customerid": null
                        }
                    }
                ]
            }
        ],
        "outcome": {
            "type": "correlate",
            "target": "eventwebhook-evp7bpraeyche8bdpijg"
        },
        "contexts": [
            {
                "id": "86241202-9aeb-4d96-85ce-2e43ecf91f64",
                "mappings": {
                    "customerid": "CUSTOMER-2"
                },
                "pendingEvents": []
            }
        ]
    },
    {
        "id": "4965e2e9-9580-47f9-ad04-e332a9599ce8",
        "createdAt": "2023-03-14T18: 46: 34.6087596",
        "lastModified": "2023-03-14T18: 46: 34.6087596",
        "activationType": "implicit",
        "conditions": [
            {
                "filters": [
                    {
                        "attributes": {
                            "source": "https: //demo.synpase.com/workflow-resume-event",
                            "type": "com.synapse.demo/workflow-resume-event"
                        },
                        "correlationMappings": {
                            "customerid": null
                        }
                    }
                ]
            }
        ],
        "outcome": {
            "type": "correlate",
            "target": "eventwebhook-ypppuxdwy0m5dc0cehhra"
        },
        "contexts": [
            {
                "id": "92889471-53e6-4433-9945-74d0f210884f",
                "mappings": {
                    "customerid": "CUSTOMER-1"
                },
                "pendingEvents": []
            }
        ]
    }
]