mirror of
https://github.com/hanzo-fi/payments.git
synced 2026-10-30 20:31:20 +00:00
fgueryandGitHub
599477b709
* chore: rename RunNextTasks workflow * feat: added a trimmed down type for connector only containing the id * feat: make runNextTask use connectorIdOnly * feat: added isScheduledForDeletion to plugin interface * feat: add scheduledForDeletion at plugin level * feat: refactor manager.Load to take a connector instead of parts of it * feat: saves the connector's scheduledFOrDeletion into the plugin * feat: use plugin.IsScheduledForDeletion instead of loading connector * fix: make e2e tests less flaky * fix: remove config from most request types * feat: (WIP) remove config from runNextTask * fix: remove unneeded method definitions in each plugin * fix: simplify RunNextFlow as anyway we need to change the parameters * fix: add a workflow compatible with previous version for runNextTasks * fix(EN-577): added back RunSendEvents as a deprecated workflow * fix(EN-577): added back RunUpdatePaymentInitiationFromPayment as a deprecated workflow * fix: fix linting directives * fix: replace old usage with refactored one * fix: added back missing deprecated activities * fix: ensure fetch_next_balance works with v3.0 code * fix: simplify a bit the versionning etc * fix: quick refactor before applying solution elsewhere * fix: add back RunSendEvents and RunNextTask child workflows where necessary for backward compatibility * fix: add back missing activity, seems like it's mandatory? * fix: run payment initiation update as child workflow when necessary * fix: missing activy call in test, refine some todos * fix: lint * fix: ensure we always reload plugins into the manager -- so exposing isScheduledForDeletion * fix: ensure we stop storing clear config in temporal * fix: make store config support plain text config * fix: some comments * fix: change runSendEvent workflow to be match what we had in 3.0 * feat: enable worker versionning to avoid workflow from new version to get executed on old code * fix: lint * fix: revert worker versionning * fix: minor code review comments * fix: add context to Encrypt & Decrypt methods * fix: improve lint, add couple tests * fix: use atomic to handle isScheduledForDeletion * fix: make create_events work on migration * fix: avoid returning channel; just send what is needed in runSendEventAsChildWorkflow and runNextTaskAsChildWorkflow * fix: remove extra call to installWorkflow on start plugin * fix: typo in error log line * fix: simplify errChan initialization * fix: clarify feature flag names
287 lines
11 KiB
Go
287 lines
11 KiB
Go
package engine_test
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"time"
|
|
|
|
"github.com/formancehq/go-libs/v3/bun/bunpaginate"
|
|
"github.com/formancehq/go-libs/v3/logging"
|
|
"github.com/formancehq/go-libs/v3/temporal"
|
|
"github.com/formancehq/payments/internal/connectors"
|
|
"github.com/formancehq/payments/internal/connectors/engine"
|
|
"github.com/formancehq/payments/internal/connectors/engine/activities"
|
|
"github.com/formancehq/payments/internal/models"
|
|
"github.com/formancehq/payments/internal/storage"
|
|
"github.com/google/uuid"
|
|
. "github.com/onsi/ginkgo/v2"
|
|
. "github.com/onsi/gomega"
|
|
"go.temporal.io/api/enums/v1"
|
|
"go.temporal.io/api/serviceerror"
|
|
"go.temporal.io/sdk/client"
|
|
sdktemporal "go.temporal.io/sdk/temporal"
|
|
"go.temporal.io/sdk/worker"
|
|
gomock "go.uber.org/mock/gomock"
|
|
)
|
|
|
|
var _ = Describe("Worker Tests", func() {
|
|
Context("on start", func() {
|
|
var (
|
|
pool *engine.WorkerPool
|
|
store *storage.MockStorage
|
|
manager *connectors.MockManager
|
|
conns []models.Connector
|
|
)
|
|
BeforeEach(func() {
|
|
ctrl := gomock.NewController(GinkgoT())
|
|
logger := logging.NewDefaultLogger(GinkgoWriter, false, false, false)
|
|
// Use NewLazyClient as worker.New() requires a properly created client
|
|
cl, err := client.NewLazyClient(client.Options{})
|
|
Expect(err).To(BeNil())
|
|
store = storage.NewMockStorage(ctrl)
|
|
manager = connectors.NewMockManager(ctrl)
|
|
|
|
pool = engine.NewWorkerPool(
|
|
logger,
|
|
"stackname",
|
|
cl,
|
|
[]temporal.DefinitionSet{},
|
|
[]temporal.DefinitionSet{},
|
|
store,
|
|
manager,
|
|
worker.Options{},
|
|
time.Second,
|
|
time.Hour,
|
|
)
|
|
// Skip schedule creation in tests since we don't have a Temporal server
|
|
pool.SetSkipScheduleCreation(true)
|
|
|
|
connID1 := models.ConnectorID{Reference: uuid.New(), Provider: "provider1"}
|
|
connID2 := models.ConnectorID{Reference: uuid.New(), Provider: "provider2"}
|
|
|
|
conns = []models.Connector{
|
|
{ConnectorBase: models.ConnectorBase{ID: connID1, Name: "abc-connector", Provider: connID1.Provider, CreatedAt: time.Now().Add(-time.Minute)}, Config: json.RawMessage(`{}`)},
|
|
{ConnectorBase: models.ConnectorBase{ID: connID2, Name: "efg-connector", Provider: connID2.Provider, CreatedAt: time.Now()}, Config: json.RawMessage(`{}`)},
|
|
}
|
|
|
|
})
|
|
|
|
It("should fail when listener fails", func(ctx SpecContext) {
|
|
expectedErr := fmt.Errorf("listener err")
|
|
store.EXPECT().ListenConnectorsChanges(gomock.Any(), gomock.Any()).Return(expectedErr)
|
|
err := pool.OnStart(ctx)
|
|
Expect(err).NotTo(BeNil())
|
|
Expect(err).To(MatchError(expectedErr))
|
|
})
|
|
|
|
It("should fail when unable to fetch connectors from storage", func(ctx SpecContext) {
|
|
store.EXPECT().ListenConnectorsChanges(gomock.Any(), gomock.Any()).Return(nil)
|
|
|
|
expectedErr := fmt.Errorf("storage err")
|
|
store.EXPECT().ConnectorsList(gomock.Any(), gomock.Any()).Return(nil, expectedErr)
|
|
err := pool.OnStart(ctx)
|
|
Expect(err).NotTo(BeNil())
|
|
Expect(err).To(MatchError(expectedErr))
|
|
})
|
|
|
|
It("should call RegisterPlugin on all connectors found", func(ctx SpecContext) {
|
|
store.EXPECT().ListenConnectorsChanges(gomock.Any(), gomock.Any()).Return(nil)
|
|
|
|
store.EXPECT().ConnectorsList(gomock.Any(), gomock.Any()).Return(&bunpaginate.Cursor[models.Connector]{
|
|
Data: conns,
|
|
}, nil)
|
|
manager.EXPECT().Load(conns[0], false, false).Return("name", json.RawMessage(`{}`), nil)
|
|
manager.EXPECT().Load(conns[1], false, false).Return("name", json.RawMessage(`{}`), nil)
|
|
err := pool.OnStart(ctx)
|
|
Expect(err).To(BeNil())
|
|
})
|
|
})
|
|
|
|
Context("createOutboxPublisherSchedule", func() {
|
|
var (
|
|
pool *engine.WorkerPool
|
|
mockClient *activities.MockClient
|
|
mockScheduleClient *activities.MockScheduleClient
|
|
mockHandle *activities.MockScheduleHandle
|
|
stackName string
|
|
|
|
pollingInterval = time.Second
|
|
cleanupInterval = time.Hour
|
|
)
|
|
|
|
BeforeEach(func() {
|
|
ctrl := gomock.NewController(GinkgoT())
|
|
logger := logging.NewDefaultLogger(GinkgoWriter, false, false, false)
|
|
stackName = "test-stack"
|
|
mockClient = activities.NewMockClient(ctrl)
|
|
mockScheduleClient = activities.NewMockScheduleClient(ctrl)
|
|
mockHandle = activities.NewMockScheduleHandle(ctrl)
|
|
store := storage.NewMockStorage(ctrl)
|
|
manager := connectors.NewMockManager(ctrl)
|
|
pool = engine.NewWorkerPool(
|
|
logger,
|
|
stackName,
|
|
mockClient,
|
|
[]temporal.DefinitionSet{},
|
|
[]temporal.DefinitionSet{},
|
|
store,
|
|
manager,
|
|
worker.Options{},
|
|
pollingInterval,
|
|
cleanupInterval,
|
|
)
|
|
// Don't skip schedule creation for these tests
|
|
pool.SetSkipScheduleCreation(false)
|
|
})
|
|
|
|
It("should successfully create schedule when it does not exist", func(ctx SpecContext) {
|
|
scheduleID := fmt.Sprintf("%s-outbox-publisher", stackName)
|
|
mockClient.EXPECT().ScheduleClient().Return(mockScheduleClient).AnyTimes()
|
|
mockScheduleClient.EXPECT().Create(ctx, gomock.Any()).Do(func(_ context.Context, opts client.ScheduleOptions) {
|
|
Expect(opts.ID).To(Equal(scheduleID))
|
|
Expect(opts.TriggerImmediately).To(BeTrue())
|
|
Expect(opts.Overlap).To(Equal(enums.SCHEDULE_OVERLAP_POLICY_SKIP))
|
|
Expect(opts.Spec.Intervals).To(HaveLen(1))
|
|
Expect(opts.Spec.Intervals[0].Every).To(Equal(pollingInterval))
|
|
//nolint:staticcheck
|
|
Expect(opts.SearchAttributes["Stack"]).To(Equal(stackName))
|
|
action, ok := opts.Action.(*client.ScheduleWorkflowAction)
|
|
Expect(ok).To(BeTrue())
|
|
Expect(action.Workflow).To(Equal("OutboxPublisher"))
|
|
Expect(action.TaskQueue).To(Equal(fmt.Sprintf("%s-default", stackName)))
|
|
}).Return(mockHandle, nil)
|
|
|
|
err := pool.CreateOutboxPublisherSchedule(ctx)
|
|
Expect(err).To(BeNil())
|
|
})
|
|
|
|
It("should return nil when schedule already exists (AlreadyExists error)", func(ctx SpecContext) {
|
|
mockClient.EXPECT().ScheduleClient().Return(mockScheduleClient).AnyTimes()
|
|
mockScheduleClient.EXPECT().Create(ctx, gomock.Any()).Return(nil, serviceerror.NewAlreadyExists("already exists"))
|
|
|
|
err := pool.CreateOutboxPublisherSchedule(ctx)
|
|
Expect(err).To(BeNil())
|
|
})
|
|
|
|
It("should return nil when initial workflow already started (WorkflowExecutionAlreadyStarted)", func(ctx SpecContext) {
|
|
mockClient.EXPECT().ScheduleClient().Return(mockScheduleClient).AnyTimes()
|
|
mockScheduleClient.EXPECT().Create(ctx, gomock.Any()).Return(nil, serviceerror.NewWorkflowExecutionAlreadyStarted("wf already started", "", ""))
|
|
|
|
err := pool.CreateOutboxPublisherSchedule(ctx)
|
|
Expect(err).To(BeNil())
|
|
})
|
|
|
|
It("should return nil when SDK reports ErrScheduleAlreadyRunning", func(ctx SpecContext) {
|
|
mockClient.EXPECT().ScheduleClient().Return(mockScheduleClient).AnyTimes()
|
|
mockScheduleClient.EXPECT().Create(ctx, gomock.Any()).Return(nil, sdktemporal.ErrScheduleAlreadyRunning)
|
|
|
|
err := pool.CreateOutboxPublisherSchedule(ctx)
|
|
Expect(err).To(BeNil())
|
|
})
|
|
|
|
It("should return error when Create fails with non-AlreadyExists error", func(ctx SpecContext) {
|
|
expectedErr := fmt.Errorf("create error")
|
|
mockClient.EXPECT().ScheduleClient().Return(mockScheduleClient).AnyTimes()
|
|
mockScheduleClient.EXPECT().Create(ctx, gomock.Any()).Return(nil, expectedErr)
|
|
|
|
err := pool.CreateOutboxPublisherSchedule(ctx)
|
|
Expect(err).NotTo(BeNil())
|
|
Expect(err.Error()).To(ContainSubstring("failed to create outbox publisher schedule"))
|
|
})
|
|
})
|
|
|
|
Context("createOutboxCleanupSchedule", func() {
|
|
var (
|
|
pool *engine.WorkerPool
|
|
mockClient *activities.MockClient
|
|
mockScheduleClient *activities.MockScheduleClient
|
|
mockHandle *activities.MockScheduleHandle
|
|
stackName string
|
|
|
|
pollingInterval = time.Second
|
|
cleanupInterval = time.Hour
|
|
)
|
|
|
|
BeforeEach(func() {
|
|
ctrl := gomock.NewController(GinkgoT())
|
|
logger := logging.NewDefaultLogger(GinkgoWriter, false, false, false)
|
|
stackName = "test-stack"
|
|
mockClient = activities.NewMockClient(ctrl)
|
|
mockScheduleClient = activities.NewMockScheduleClient(ctrl)
|
|
mockHandle = activities.NewMockScheduleHandle(ctrl)
|
|
store := storage.NewMockStorage(ctrl)
|
|
manager := connectors.NewMockManager(ctrl)
|
|
pool = engine.NewWorkerPool(
|
|
logger,
|
|
stackName,
|
|
mockClient,
|
|
[]temporal.DefinitionSet{},
|
|
[]temporal.DefinitionSet{},
|
|
store,
|
|
manager,
|
|
worker.Options{},
|
|
pollingInterval,
|
|
cleanupInterval,
|
|
)
|
|
// Don't skip schedule creation for these tests
|
|
pool.SetSkipScheduleCreation(false)
|
|
})
|
|
|
|
It("should successfully create schedule when it does not exist", func(ctx SpecContext) {
|
|
scheduleID := fmt.Sprintf("%s-outbox-cleanup", stackName)
|
|
mockClient.EXPECT().ScheduleClient().Return(mockScheduleClient).AnyTimes()
|
|
mockScheduleClient.EXPECT().Create(ctx, gomock.Any()).Do(func(_ context.Context, opts client.ScheduleOptions) {
|
|
Expect(opts.ID).To(Equal(scheduleID))
|
|
Expect(opts.TriggerImmediately).To(BeTrue())
|
|
Expect(opts.Overlap).To(Equal(enums.SCHEDULE_OVERLAP_POLICY_SKIP))
|
|
Expect(opts.Spec.Intervals).To(HaveLen(1))
|
|
Expect(opts.Spec.Intervals[0].Every).To(Equal(cleanupInterval))
|
|
//nolint:staticcheck
|
|
Expect(opts.SearchAttributes["Stack"]).To(Equal(stackName))
|
|
action, ok := opts.Action.(*client.ScheduleWorkflowAction)
|
|
Expect(ok).To(BeTrue())
|
|
Expect(action.Workflow).To(Equal("OutboxCleanup"))
|
|
Expect(action.TaskQueue).To(Equal(fmt.Sprintf("%s-default", stackName)))
|
|
}).Return(mockHandle, nil)
|
|
|
|
err := pool.CreateOutboxCleanupSchedule(ctx)
|
|
Expect(err).To(BeNil())
|
|
})
|
|
|
|
It("should return nil when schedule already exists (AlreadyExists error)", func(ctx SpecContext) {
|
|
mockClient.EXPECT().ScheduleClient().Return(mockScheduleClient).AnyTimes()
|
|
mockScheduleClient.EXPECT().Create(ctx, gomock.Any()).Return(nil, serviceerror.NewAlreadyExists("already exists"))
|
|
|
|
err := pool.CreateOutboxCleanupSchedule(ctx)
|
|
Expect(err).To(BeNil())
|
|
})
|
|
|
|
It("should return nil when initial workflow already started (WorkflowExecutionAlreadyStarted)", func(ctx SpecContext) {
|
|
mockClient.EXPECT().ScheduleClient().Return(mockScheduleClient).AnyTimes()
|
|
mockScheduleClient.EXPECT().Create(ctx, gomock.Any()).Return(nil, serviceerror.NewWorkflowExecutionAlreadyStarted("wf already started", "", ""))
|
|
|
|
err := pool.CreateOutboxCleanupSchedule(ctx)
|
|
Expect(err).To(BeNil())
|
|
})
|
|
|
|
It("should return nil when SDK reports ErrScheduleAlreadyRunning", func(ctx SpecContext) {
|
|
mockClient.EXPECT().ScheduleClient().Return(mockScheduleClient).AnyTimes()
|
|
mockScheduleClient.EXPECT().Create(ctx, gomock.Any()).Return(nil, sdktemporal.ErrScheduleAlreadyRunning)
|
|
|
|
err := pool.CreateOutboxCleanupSchedule(ctx)
|
|
Expect(err).To(BeNil())
|
|
})
|
|
|
|
It("should return error when Create fails with non-AlreadyExists error", func(ctx SpecContext) {
|
|
expectedErr := fmt.Errorf("create error")
|
|
mockClient.EXPECT().ScheduleClient().Return(mockScheduleClient).AnyTimes()
|
|
mockScheduleClient.EXPECT().Create(ctx, gomock.Any()).Return(nil, expectedErr)
|
|
|
|
err := pool.CreateOutboxCleanupSchedule(ctx)
|
|
Expect(err).NotTo(BeNil())
|
|
Expect(err.Error()).To(ContainSubstring("failed to create outbox cleanup schedule"))
|
|
})
|
|
})
|
|
})
|