feat(orchestration): add some events (#1286)

components/ledger/libs/publish/module.go

@@ -6,6 +6,7 @@ import (
 	"github.com/ThreeDotsLabs/watermill/message"
 	"github.com/ThreeDotsLabs/watermill/pubsub/gochannel"
 	"go.uber.org/fx"
+	"sync"
 )
 
 func newGoChannel() *gochannel.GoChannel {
@@ -95,3 +96,49 @@ func NewTopicMapperPublisherDecorator(publisher message.Publisher, topics map[st
 		topics:    topics,
 	}
 }
+
+type noOpPublisher struct {
+}
+
+func (n noOpPublisher) Publish(topic string, messages ...*message.Message) error {
+	return nil
+}
+
+func (n noOpPublisher) Close() error {
+	return nil
+}
+
+var NoOpPublisher message.Publisher = &noOpPublisher{}
+
+type memoryPublisher struct {
+	sync.Mutex
+	messages map[string][]*message.Message
+}
+
+func (m *memoryPublisher) Publish(topic string, messages ...*message.Message) error {
+	m.Lock()
+	defer m.Unlock()
+
+	m.messages[topic] = append(m.messages[topic], messages...)
+	return nil
+}
+
+func (m *memoryPublisher) Close() error {
+	m.Lock()
+	defer m.Unlock()
+
+	m.messages = map[string][]*message.Message{}
+	return nil
+}
+
+func (m *memoryPublisher) AllMessages() map[string][]*message.Message {
+	return m.messages
+}
+
+var _ message.Publisher = (*memoryPublisher)(nil)
+
+func InMemory() *memoryPublisher {
+	return &memoryPublisher{
+		messages: map[string][]*message.Message{},
+	}
+}

ee/orchestration/internal/triggers/activity.go

@@ -5,6 +5,8 @@ import (
 	"encoding/json"
 	"strings"
 
+	"github.com/ThreeDotsLabs/watermill/message"
+	"github.com/formancehq/orchestration/pkg/events"
 	sharedlogging "github.com/formancehq/stack/libs/go-libs/logging"
 
 	"go.temporal.io/sdk/temporal"
@@ -21,6 +23,7 @@ type Activities struct {
 	db                  *bun.DB
 	manager             *workflow.WorkflowManager
 	expressionEvaluator *expressionEvaluator
+	publisher           message.Publisher
 }
 
 func (a Activities) processTrigger(ctx context.Context, request ProcessEventRequest, trigger Trigger) bool {
@@ -74,7 +77,7 @@ func (a Activities) ListTriggers(ctx context.Context, request ProcessEventReques
 	return ret, nil
 }
 
-func (a Activities) ProcessTrigger(ctx context.Context, trigger Trigger, request ProcessEventRequest) error {
+func (a Activities) ProcessTrigger(ctx context.Context, trigger Trigger, request ProcessEventRequest) (*Occurrence, error) {
 
 	span := trace.SpanFromContext(ctx)
 	var (
@@ -95,7 +98,7 @@ func (a Activities) ProcessTrigger(ctx context.Context, trigger Trigger, request
 
 		instance, triggerError := a.manager.RunWorkflow(ctx, trigger.WorkflowID, evaluated)
 		if triggerError != nil {
-			return triggerError
+			return nil, triggerError
 		}
 
 		occurrence.WorkflowInstanceID = pointer.For(instance.ID)
@@ -113,16 +116,36 @@ func (a Activities) ProcessTrigger(ctx context.Context, trigger Trigger, request
 		sharedlogging.FromContext(ctx).Errorf("unable to save trigger occurrence: %s", err)
 	}
 
-	return triggerError
+	return &occurrence, nil
 }
 
-func NewActivities(db *bun.DB, manager *workflow.WorkflowManager, expressionEvaluator *expressionEvaluator) Activities {
+func (a Activities) SendEventForTriggerTermination(ctx context.Context, occurrence Occurrence) error {
+	if occurrence.Error == nil || *occurrence.Error == "" {
+		return a.publisher.Publish(events.SucceededTrigger,
+			events.NewMessage(ctx, events.SucceededTrigger, events.SucceededTriggerPayload{
+				ID:      occurrence.TriggerID,
+				EventID: occurrence.EventID,
+			}))
+	} else {
+		return a.publisher.Publish(events.FailedTrigger,
+			events.NewMessage(ctx, events.FailedTrigger, events.FailedTriggerPayload{
+				ID:      occurrence.TriggerID,
+				Error:   *occurrence.Error,
+				EventID: occurrence.EventID,
+			}))
+	}
+}
+
+func NewActivities(db *bun.DB, manager *workflow.WorkflowManager,
+	expressionEvaluator *expressionEvaluator, publisher message.Publisher) Activities {
 	return Activities{
 		db:                  db,
 		manager:             manager,
 		expressionEvaluator: expressionEvaluator,
+		publisher:           publisher,
 	}
 }
 
 var ProcessEventActivity = Activities{}.ProcessTrigger
+var SendEventForTriggerTermination = Activities{}.SendEventForTriggerTermination
 var ListTriggersActivity = Activities{}.ListTriggers

ee/orchestration/internal/triggers/module.go

@@ -24,8 +24,9 @@ func NewModule(taskQueue string) fx.Option {
 			}, fx.As(new(any)), fx.ResultTags(`group:"workflows"`)),
 		),
 		fx.Provide(
-			fx.Annotate(func(db *bun.DB, manager *workflow.WorkflowManager, expressionEvaluator *expressionEvaluator) Activities {
-				return NewActivities(db, manager, expressionEvaluator)
+			fx.Annotate(func(db *bun.DB, manager *workflow.WorkflowManager,
+				expressionEvaluator *expressionEvaluator, publisher message.Publisher) Activities {
+				return NewActivities(db, manager, expressionEvaluator, publisher)
 			}, fx.As(new(any)), fx.ResultTags(`group:"activities"`)),
 		),
 	)

ee/orchestration/internal/triggers/workflow_trigger.go

@@ -3,9 +3,6 @@ package triggers
 import (
 	"time"
 
-	"github.com/pkg/errors"
-	"go.temporal.io/sdk/temporal"
-
 	"github.com/formancehq/stack/libs/go-libs/publish"
 	"github.com/uptrace/bun"
 	temporalworkflow "go.temporal.io/sdk/workflow"
@@ -36,19 +33,28 @@ func (w triggerWorkflow) RunTrigger(ctx temporalworkflow.Context, req ProcessEve
 	}
 
 	for _, trigger := range triggers {
+		occurrence := &Occurrence{}
 		err := temporalworkflow.ExecuteActivity(
 			temporalworkflow.WithActivityOptions(ctx, temporalworkflow.ActivityOptions{
 				StartToCloseTimeout: 10 * time.Second,
 			}),
 			ProcessEventActivity,
 			trigger,
 			req,
+		).Get(ctx, occurrence)
+		if err != nil {
+			return err
+		}
+
+		err = temporalworkflow.ExecuteActivity(
+			temporalworkflow.WithActivityOptions(ctx, temporalworkflow.ActivityOptions{
+				StartToCloseTimeout: 10 * time.Second,
+			}),
+			SendEventForTriggerTermination,
+			occurrence,
 		).Get(ctx, nil)
 		if err != nil {
-			applicationError := &temporal.ApplicationError{}
-			if !errors.As(err, &applicationError) {
-				return err
-			}
+			return err
 		}
 	}
 

ee/orchestration/internal/triggers/workflow_trigger_test.go

@@ -38,6 +38,10 @@ func TestWorkflow(t *testing.T) {
 	env.
 		OnActivity(ProcessEventActivity, mock.Anything, trigger, req).
 		Once().
+		Return(&Occurrence{}, nil)
+	env.
+		OnActivity(SendEventForTriggerTermination, mock.Anything, mock.Anything).
+		Once().
 		Return(nil)
 
 	env.ExecuteWorkflow(RunTrigger, req)

ee/orchestration/internal/workflow/activities.go

@@ -0,0 +1,37 @@
+package workflow
+
+import (
+	"context"
+
+	"github.com/ThreeDotsLabs/watermill/message"
+	"github.com/formancehq/orchestration/pkg/events"
+)
+
+type Activities struct {
+	publisher message.Publisher
+}
+
+func (a Activities) SendWorkflowTerminationEvent(ctx context.Context, instance Instance) error {
+	if instance.Error == "" {
+		return a.publisher.Publish(events.SucceededWorkflow,
+			events.NewMessage(ctx, events.SucceededWorkflow, events.SucceededWorkflowPayload{
+				ID:         instance.WorkflowID,
+				InstanceID: instance.ID,
+			}))
+	} else {
+		return a.publisher.Publish(events.FailedWorkflow,
+			events.NewMessage(ctx, events.FailedWorkflow, events.FailedWorkflowPayload{
+				ID:         instance.WorkflowID,
+				InstanceID: instance.ID,
+				Error:      instance.Error,
+			}))
+	}
+}
+
+var SendWorkflowTerminationEventActivity = (&Activities{}).SendWorkflowTerminationEvent
+
+func NewActivities(publisher message.Publisher) Activities {
+	return Activities{
+		publisher: publisher,
+	}
+}

ee/orchestration/internal/workflow/activities_test.go

@@ -0,0 +1,21 @@
+package workflow
+
+import (
+	"testing"
+
+	"github.com/formancehq/stack/libs/go-libs/publish"
+	"github.com/stretchr/testify/require"
+	"go.temporal.io/sdk/testsuite"
+)
+
+func TestActivities(t *testing.T) {
+	publisher := publish.InMemory()
+	activities := NewActivities(publisher)
+
+	testSuite := &testsuite.WorkflowTestSuite{}
+	env := testSuite.NewTestActivityEnvironment()
+	env.RegisterActivity(activities.SendWorkflowTerminationEvent)
+	_, err := env.ExecuteActivity(SendWorkflowTerminationEventActivity, NewInstance("xxx"))
+	require.NoError(t, err)
+	require.NotEmpty(t, publisher.AllMessages())
+}

ee/orchestration/internal/workflow/input.go

@@ -2,6 +2,7 @@ package workflow
 
 import (
 	"context"
+	"time"
 
 	"github.com/uptrace/bun"
 	"go.temporal.io/sdk/workflow"
@@ -27,5 +28,17 @@ func (i Input) run(ctx workflow.Context, db *bun.DB) error {
 		Exec(context.Background()); dbErr != nil {
 		workflow.GetLogger(ctx).Error("error updating instance into database", "error", dbErr)
 	}
+
+	err = workflow.ExecuteActivity(
+		workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
+			StartToCloseTimeout: 10 * time.Second,
+		}),
+		SendWorkflowTerminationEventActivity,
+		instance,
+	).Get(ctx, nil)
+	if err != nil {
+		return err
+	}
+
 	return err
 }

ee/orchestration/internal/workflow/manager_test.go

@@ -4,6 +4,8 @@ import (
 	"context"
 	"testing"
 
+	"github.com/stretchr/testify/mock"
+
 	"github.com/formancehq/stack/libs/go-libs/logging"
 
 	"github.com/formancehq/stack/libs/go-libs/bun/bunconnect"
@@ -46,6 +48,7 @@ func TestConfig(t *testing.T) {
 	workflows := NewWorkflows(db)
 	env.RegisterWorkflow(stages.RunNoOp)
 	env.RegisterWorkflow(workflows.Run)
+	env.OnActivity((&Activities{}).SendWorkflowTerminationEvent, mock.Anything, mock.Anything).Return(nil)
 	mockClient := &mockTemporalClient{env: env, workflows: workflows}
 	manager := NewManager(db, mockClient, "default")
 

ee/orchestration/internal/workflow/module.go

@@ -15,6 +15,7 @@ func NewModule(taskQueue string) fx.Option {
 		}),
 		fx.Provide(fx.Annotate(NewWorkflows, fx.ResultTags(`group:"workflows"`), fx.As(new(any)))),
 		fx.Provide(fx.Annotate(activities.New, fx.ResultTags(`group:"activities"`), fx.As(new(any)))),
+		fx.Provide(fx.Annotate(NewActivities, fx.ResultTags(`group:"activities"`), fx.As(new(any)))),
 	}
 
 	for _, schema := range stages.All() {

ee/orchestration/pkg/events/events.go

@@ -0,0 +1,48 @@
+package events
+
+import (
+	"context"
+	"time"
+
+	"github.com/ThreeDotsLabs/watermill/message"
+	"github.com/formancehq/stack/libs/go-libs/publish"
+)
+
+const (
+	SucceededWorkflow = "SUCCEEDED_WORKFLOW"
+	FailedWorkflow    = "FAILED_WORKFLOW"
+	SucceededTrigger  = "SUCCEEDED_TRIGGER"
+	FailedTrigger     = "FAILED_TRIGGER"
+)
+
+type SucceededWorkflowPayload struct {
+	ID         string `json:"id"`
+	InstanceID string `json:"instanceID"`
+}
+
+type FailedWorkflowPayload struct {
+	ID         string `json:"id"`
+	InstanceID string `json:"instanceID"`
+	Error      string `json:"error"`
+}
+
+type SucceededTriggerPayload struct {
+	ID      string `json:"id"`
+	EventID string `json:"eventID"`
+}
+
+type FailedTriggerPayload struct {
+	ID      string `json:"id"`
+	EventID string `json:"eventID"`
+	Error   string `json:"error"`
+}
+
+func NewMessage(ctx context.Context, mtype string, payload any) *message.Message {
+	return publish.NewMessage(ctx, publish.EventMessage{
+		Date:    time.Now(),
+		App:     "orchestration",
+		Version: "v2",
+		Type:    mtype,
+		Payload: payload,
+	})
+}

libs/events/services/orchestration/v2.0.0/FAILED_TRIGGER.yaml

@@ -0,0 +1,12 @@
+type: object
+properties:
+  id:
+    type: string
+  eventID:
+    type: string
+  error:
+    type: string
+required:
+  - id
+  - eventID
+  - error

libs/events/services/orchestration/v2.0.0/FAILED_WORKFLOW.yaml

@@ -0,0 +1,12 @@
+type: object
+properties:
+  id:
+    type: string
+  instanceID:
+    type: string
+  error:
+    type: string
+required:
+  - id
+  - instanceID
+  - error

libs/events/services/orchestration/v2.0.0/SUCCEEDED_TRIGGER.yaml

@@ -0,0 +1,9 @@
+type: object
+properties:
+  id:
+    type: string
+  eventID:
+    type: string
+required:
+  - id
+  - eventID