mirror of
https://github.com/hanzo-fi/payments.git
synced 2026-10-30 20:31:20 +00:00
fgueryandGitHub
a63e3a20a8
* feat(EN-214): add outbox table migration * feat(EN-214): add outbox models and storage layer * feat(EN-214): implement outbox publisher activity * feat(EN-214): implement outbox schedule activity * feat(EN-214): add outbox publisher workflow * feat(EN-214): migrate accounts to outbox pattern * feat(EN-214): migrate balances to outbox pattern * feat(EN-214): migrate payments to outbox pattern * feat(EN-214): migrate bank accounts to outbox pattern * feat(EN-214): migrate pools to outbox pattern * feat(EN-214): migrate payment initiations to outbox pattern * feat(EN-214): migrate PSU and connections to outbox pattern * feat(EN-214): migrate connectors to outbox pattern * feat(EN-214): remove legacy send_events workflow * feat(EN-214): update workflows to use outbox pattern * feat(EN-214): improve outbox idempotency and error handling * feat(EN-214): update storage tasks and tests * test(EN-214): update e2e tests for outbox pattern * test(EN-214): clean up workflow tests * chore(EN-214): update docker compose configuration * feat(EN-214): add e2e run config in ide * fix: add back docker-compose-dev run config * chore: improve connector uninstall comments & remove connector id being passed * fix: ensure events are not pushed if balances are not created * fix: ensure err is not overwritten when using rollback * fix: ensure we use v3 provider format * chore: apply the nitpick comments from coderabbit * chore: consistentely use e() error handling * feat: optionally disable publisher schedule to make e2e test more consistent * fix: improve error handling on uninstall connector * avoid using custom payload and rely on existing interfaces; also avoid using different types for events * fix: remove unecessary unmarshalling * fix: ensure that rollbackOnTxError picks up errors on tx.Commit() * fix: add extra validation on iban & account number in the test * test: cover marshall/unmarshalling of event payload * fix: increase test timeout * try to make test more resiliant * fix: wrong pointer use in loop * fix: error formatting issue * fix: move oubtox schedule creation out of an activity and do it in the worker itself * feat: OutboxEventDeleteAndSend as batch * change outbox_event to have id only instead of idempotency key * make e2e test less brittle, save eventID as idempotency key in publication * feat: make outbox_event keep an history * feat(EN-214): added a schedule to cleanup outbox events * test: make cleanup actually clean * ignore linting rules for now * feat: do not add psu metadata to event anymore as it's irrelevant * fix: create local variable instead of relying on pointers, even if not 100% necessary * fix: remove temporal schedule describe calls
471 lines
14 KiB
Go
471 lines
14 KiB
Go
package activities_test
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"time"
|
|
|
|
"github.com/formancehq/go-libs/v3/logging"
|
|
"github.com/formancehq/payments/internal/connectors/engine/activities"
|
|
internalevents "github.com/formancehq/payments/internal/events"
|
|
"github.com/formancehq/payments/internal/models"
|
|
"github.com/formancehq/payments/internal/storage"
|
|
"github.com/formancehq/payments/pkg/events"
|
|
"github.com/google/uuid"
|
|
. "github.com/onsi/ginkgo/v2"
|
|
. "github.com/onsi/gomega"
|
|
"go.uber.org/mock/gomock"
|
|
)
|
|
|
|
var _ = Describe("OutboxPublishPendingEvents", func() {
|
|
var (
|
|
act activities.Activities
|
|
s *storage.MockStorage
|
|
evts *internalevents.Events
|
|
mockPublisher *activities.MockPublisher
|
|
logger = logging.NewDefaultLogger(GinkgoWriter, true, false, false)
|
|
)
|
|
|
|
BeforeEach(func() {
|
|
ctrl := gomock.NewController(GinkgoT())
|
|
s = storage.NewMockStorage(ctrl)
|
|
mockPublisher = activities.NewMockPublisher(ctrl)
|
|
evts = internalevents.New(mockPublisher, "http://localhost")
|
|
act = activities.New(logger, nil, s, evts, nil, 0)
|
|
})
|
|
|
|
Context("when polling pending outbox events", func() {
|
|
It("successfully publishes events", func(ctx SpecContext) {
|
|
connectorID := &models.ConnectorID{
|
|
Provider: "stripe",
|
|
Reference: uuid.New(),
|
|
}
|
|
|
|
testEvents := []models.OutboxEvent{
|
|
{
|
|
ID: models.EventID{
|
|
EventIdempotencyKey: "account.saved:acc_123",
|
|
ConnectorID: connectorID,
|
|
},
|
|
EventType: events.EventTypeSavedAccounts,
|
|
EntityID: "acc_123",
|
|
Payload: json.RawMessage(`{"id":"acc_123","name":"Test Account"}`),
|
|
CreatedAt: time.Now().UTC(),
|
|
Status: models.OUTBOX_STATUS_PENDING,
|
|
ConnectorID: connectorID,
|
|
RetryCount: 0,
|
|
},
|
|
}
|
|
|
|
// Poll events
|
|
s.EXPECT().
|
|
OutboxEventsPollPending(ctx, 100).
|
|
Return(testEvents, nil)
|
|
|
|
// Publish event (expect generic topic from events.Publish)
|
|
mockPublisher.EXPECT().
|
|
Publish(gomock.Any(), gomock.Any()).
|
|
Return(nil)
|
|
|
|
// Mark as processed and record sent atomically (batch)
|
|
s.EXPECT().
|
|
OutboxEventsMarkProcessedAndRecordSent(ctx, gomock.Any(), gomock.Any()).
|
|
Do(func(_ context.Context, eventIDs []models.EventID, eventsSent []models.EventSent) {
|
|
Expect(eventIDs).To(HaveLen(1))
|
|
Expect(eventIDs[0]).To(Equal(testEvents[0].ID))
|
|
Expect(eventsSent).To(HaveLen(1))
|
|
Expect(eventsSent[0].ConnectorID).To(Equal(connectorID))
|
|
Expect(eventsSent[0].ID).To(Equal(testEvents[0].ID))
|
|
}).
|
|
Return(nil)
|
|
|
|
err := act.OutboxPublishPendingEvents(ctx, 100)
|
|
Expect(err).To(BeNil())
|
|
})
|
|
|
|
It("returns nil when no events are pending", func(ctx SpecContext) {
|
|
s.EXPECT().
|
|
OutboxEventsPollPending(ctx, 100).
|
|
Return([]models.OutboxEvent{}, nil)
|
|
|
|
err := act.OutboxPublishPendingEvents(ctx, 100)
|
|
Expect(err).To(BeNil())
|
|
})
|
|
|
|
It("handles poll error", func(ctx SpecContext) {
|
|
expectedErr := errors.New("database error")
|
|
s.EXPECT().
|
|
OutboxEventsPollPending(ctx, 100).
|
|
Return(nil, expectedErr)
|
|
|
|
err := act.OutboxPublishPendingEvents(ctx, 100)
|
|
Expect(err).ToNot(BeNil())
|
|
Expect(err.Error()).To(ContainSubstring("failed to poll pending outbox events"))
|
|
})
|
|
|
|
It("marks event as failed after max retries", func(ctx SpecContext) {
|
|
connectorID := &models.ConnectorID{
|
|
Provider: "stripe",
|
|
Reference: uuid.New(),
|
|
}
|
|
|
|
testEvents := []models.OutboxEvent{
|
|
{
|
|
ID: models.EventID{
|
|
EventIdempotencyKey: "account.saved:acc_123",
|
|
ConnectorID: connectorID,
|
|
},
|
|
EventType: "account.saved",
|
|
EntityID: "acc_123",
|
|
Payload: json.RawMessage(`{"id":"acc_123","name":"Test Account"}`),
|
|
CreatedAt: time.Now().UTC(),
|
|
Status: models.OUTBOX_STATUS_PENDING,
|
|
ConnectorID: connectorID,
|
|
RetryCount: models.MaxOutboxRetries, // Already at max retries
|
|
},
|
|
}
|
|
|
|
// Poll events
|
|
s.EXPECT().
|
|
OutboxEventsPollPending(ctx, 100).
|
|
Return(testEvents, nil)
|
|
|
|
// Publish fails
|
|
publishErr := errors.New("publish error")
|
|
mockPublisher.EXPECT().
|
|
Publish(gomock.Any(), gomock.Any()).
|
|
Return(publishErr)
|
|
|
|
// Mark as failed (exceeds max retries)
|
|
s.EXPECT().
|
|
OutboxEventsMarkFailed(ctx, testEvents[0].ID, models.MaxOutboxRetries+1, publishErr).
|
|
Return(nil)
|
|
|
|
err := act.OutboxPublishPendingEvents(ctx, 100)
|
|
Expect(err).To(BeNil()) // Activity doesn't fail, just marks event as failed
|
|
})
|
|
|
|
It("retries failed event", func(ctx SpecContext) {
|
|
connectorID := &models.ConnectorID{
|
|
Provider: "stripe",
|
|
Reference: uuid.New(),
|
|
}
|
|
|
|
testEvents := []models.OutboxEvent{
|
|
{
|
|
ID: models.EventID{
|
|
EventIdempotencyKey: "account.saved:acc_123",
|
|
ConnectorID: connectorID,
|
|
},
|
|
EventType: "account.saved",
|
|
EntityID: "acc_123",
|
|
Payload: json.RawMessage(`{"id":"acc_123","name":"Test Account"}`),
|
|
CreatedAt: time.Now().UTC(),
|
|
Status: models.OUTBOX_STATUS_PENDING,
|
|
ConnectorID: connectorID,
|
|
RetryCount: 2, // Below max retries
|
|
},
|
|
}
|
|
|
|
// Poll events
|
|
s.EXPECT().
|
|
OutboxEventsPollPending(ctx, 100).
|
|
Return(testEvents, nil)
|
|
|
|
// Publish fails
|
|
publishErr := errors.New("publish error")
|
|
mockPublisher.EXPECT().
|
|
Publish(gomock.Any(), gomock.Any()).
|
|
Return(publishErr)
|
|
|
|
// Mark as pending for retry
|
|
s.EXPECT().
|
|
OutboxEventsMarkFailed(ctx, testEvents[0].ID, 3, publishErr).
|
|
Return(nil)
|
|
|
|
err := act.OutboxPublishPendingEvents(ctx, 100)
|
|
Expect(err).To(BeNil()) // Activity doesn't fail, just marks event for retry
|
|
})
|
|
|
|
It("publishes even unknown event type", func(ctx SpecContext) {
|
|
connectorID := &models.ConnectorID{
|
|
Provider: "stripe",
|
|
Reference: uuid.New(),
|
|
}
|
|
|
|
testEvents := []models.OutboxEvent{
|
|
{
|
|
ID: models.EventID{
|
|
EventIdempotencyKey: "unknown.event:some_id",
|
|
ConnectorID: connectorID,
|
|
},
|
|
EventType: "unknown.event",
|
|
EntityID: "some_id",
|
|
Payload: json.RawMessage(`{}`),
|
|
CreatedAt: time.Now().UTC(),
|
|
Status: models.OUTBOX_STATUS_PENDING,
|
|
ConnectorID: connectorID,
|
|
RetryCount: 0,
|
|
},
|
|
}
|
|
|
|
// Poll events
|
|
s.EXPECT().
|
|
OutboxEventsPollPending(ctx, 100).
|
|
Return(testEvents, nil)
|
|
|
|
mockPublisher.EXPECT().
|
|
Publish(gomock.Any(), gomock.Any()).
|
|
Return(nil)
|
|
|
|
// Mark as processed and record sent atomically (batch)
|
|
s.EXPECT().
|
|
OutboxEventsMarkProcessedAndRecordSent(ctx, gomock.Any(), gomock.Any()).
|
|
Do(func(_ context.Context, eventIDs []models.EventID, eventsSent []models.EventSent) {
|
|
Expect(eventIDs).To(HaveLen(1))
|
|
Expect(eventIDs[0]).To(Equal(testEvents[0].ID))
|
|
Expect(eventsSent).To(HaveLen(1))
|
|
Expect(eventsSent[0].ConnectorID).To(Equal(connectorID))
|
|
Expect(eventsSent[0].ID).To(Equal(testEvents[0].ID))
|
|
}).
|
|
Return(nil)
|
|
|
|
err := act.OutboxPublishPendingEvents(ctx, 100)
|
|
Expect(err).To(BeNil())
|
|
})
|
|
|
|
It("handles publish error while marking failed", func(ctx SpecContext) {
|
|
connectorID := &models.ConnectorID{
|
|
Provider: "stripe",
|
|
Reference: uuid.New(),
|
|
}
|
|
|
|
testEvents := []models.OutboxEvent{
|
|
{
|
|
EventType: "account.saved",
|
|
EntityID: "acc_123",
|
|
Payload: json.RawMessage(`{"id":"acc_123","name":"Test Account"}`),
|
|
CreatedAt: time.Now().UTC(),
|
|
Status: models.OUTBOX_STATUS_PENDING,
|
|
ConnectorID: connectorID,
|
|
RetryCount: 0,
|
|
},
|
|
}
|
|
|
|
// Poll events
|
|
s.EXPECT().
|
|
OutboxEventsPollPending(ctx, 100).
|
|
Return(testEvents, nil)
|
|
|
|
// Publish fails
|
|
publishErr := errors.New("publish error")
|
|
mockPublisher.EXPECT().
|
|
Publish(gomock.Any(), gomock.Any()).
|
|
Return(publishErr)
|
|
|
|
// Mark as failed also fails
|
|
markErr := errors.New("mark failed error")
|
|
s.EXPECT().
|
|
OutboxEventsMarkFailed(ctx, testEvents[0].ID, 1, publishErr).
|
|
Return(markErr)
|
|
|
|
err := act.OutboxPublishPendingEvents(ctx, 100)
|
|
Expect(err).ToNot(BeNil())
|
|
Expect(err.Error()).To(ContainSubstring("failed to update event retry count"))
|
|
})
|
|
|
|
It("handles delete and record sent error", func(ctx SpecContext) {
|
|
connectorID := &models.ConnectorID{
|
|
Provider: "stripe",
|
|
Reference: uuid.New(),
|
|
}
|
|
|
|
testEvents := []models.OutboxEvent{
|
|
{
|
|
ID: models.EventID{
|
|
EventIdempotencyKey: "account.saved:acc_123",
|
|
ConnectorID: connectorID,
|
|
},
|
|
EventType: "account.saved",
|
|
EntityID: "acc_123",
|
|
Payload: json.RawMessage(`{"id":"acc_123","name":"Test Account"}`),
|
|
CreatedAt: time.Now().UTC(),
|
|
Status: models.OUTBOX_STATUS_PENDING,
|
|
ConnectorID: connectorID,
|
|
RetryCount: 0,
|
|
},
|
|
}
|
|
|
|
// Poll events
|
|
s.EXPECT().
|
|
OutboxEventsPollPending(ctx, 100).
|
|
Return(testEvents, nil)
|
|
|
|
// Publish succeeds
|
|
mockPublisher.EXPECT().
|
|
Publish(gomock.Any(), gomock.Any()).
|
|
Return(nil)
|
|
|
|
// Mark as processed and record sent fails
|
|
markErr := errors.New("mark processed error")
|
|
s.EXPECT().
|
|
OutboxEventsMarkProcessedAndRecordSent(ctx, gomock.Any(), gomock.Any()).
|
|
Return(markErr)
|
|
|
|
err := act.OutboxPublishPendingEvents(ctx, 100)
|
|
Expect(err).ToNot(BeNil())
|
|
Expect(err.Error()).To(ContainSubstring("failed to mark outbox events as processed and record sent"))
|
|
})
|
|
|
|
It("processes multiple events in batch", func(ctx SpecContext) {
|
|
connectorID := &models.ConnectorID{
|
|
Provider: "stripe",
|
|
Reference: uuid.New(),
|
|
}
|
|
|
|
testEvents := []models.OutboxEvent{
|
|
{
|
|
ID: models.EventID{
|
|
EventIdempotencyKey: "account.saved:acc_123",
|
|
ConnectorID: connectorID,
|
|
},
|
|
EventType: events.EventTypeSavedAccounts,
|
|
EntityID: "acc_123",
|
|
Payload: json.RawMessage(`{"id":"acc_123","name":"Test Account 1"}`),
|
|
CreatedAt: time.Now().UTC(),
|
|
Status: models.OUTBOX_STATUS_PENDING,
|
|
ConnectorID: connectorID,
|
|
RetryCount: 0,
|
|
},
|
|
{
|
|
ID: models.EventID{
|
|
EventIdempotencyKey: "account.saved:acc_456",
|
|
ConnectorID: connectorID,
|
|
},
|
|
EventType: events.EventTypeSavedAccounts,
|
|
EntityID: "acc_456",
|
|
Payload: json.RawMessage(`{"id":"acc_456","name":"Test Account 2"}`),
|
|
CreatedAt: time.Now().UTC(),
|
|
Status: models.OUTBOX_STATUS_PENDING,
|
|
ConnectorID: connectorID,
|
|
RetryCount: 0,
|
|
},
|
|
{
|
|
ID: models.EventID{
|
|
EventIdempotencyKey: "account.saved:acc_789",
|
|
ConnectorID: connectorID,
|
|
},
|
|
EventType: events.EventTypeSavedAccounts,
|
|
EntityID: "acc_789",
|
|
Payload: json.RawMessage(`{"id":"acc_789","name":"Test Account 3"}`),
|
|
CreatedAt: time.Now().UTC(),
|
|
Status: models.OUTBOX_STATUS_PENDING,
|
|
ConnectorID: connectorID,
|
|
RetryCount: 0,
|
|
},
|
|
}
|
|
|
|
// Poll events
|
|
s.EXPECT().
|
|
OutboxEventsPollPending(ctx, 100).
|
|
Return(testEvents, nil)
|
|
|
|
// Publish all events successfully
|
|
mockPublisher.EXPECT().
|
|
Publish(gomock.Any(), gomock.Any()).
|
|
Return(nil).
|
|
Times(3)
|
|
|
|
// Batch mark as processed and record sent
|
|
s.EXPECT().
|
|
OutboxEventsMarkProcessedAndRecordSent(ctx, gomock.Any(), gomock.Any()).
|
|
Do(func(_ context.Context, eventIDs []models.EventID, eventsSent []models.EventSent) {
|
|
Expect(eventIDs).To(HaveLen(3))
|
|
Expect(eventsSent).To(HaveLen(3))
|
|
Expect(eventIDs[0]).To(Equal(testEvents[0].ID))
|
|
Expect(eventIDs[1]).To(Equal(testEvents[1].ID))
|
|
Expect(eventIDs[2]).To(Equal(testEvents[2].ID))
|
|
Expect(eventsSent[0].ID).To(Equal(testEvents[0].ID))
|
|
Expect(eventsSent[1].ID).To(Equal(testEvents[1].ID))
|
|
Expect(eventsSent[2].ID).To(Equal(testEvents[2].ID))
|
|
}).
|
|
Return(nil)
|
|
|
|
err := act.OutboxPublishPendingEvents(ctx, 100)
|
|
Expect(err).To(BeNil())
|
|
})
|
|
|
|
It("only batches successful events", func(ctx SpecContext) {
|
|
connectorID := &models.ConnectorID{
|
|
Provider: "stripe",
|
|
Reference: uuid.New(),
|
|
}
|
|
|
|
testEvents := []models.OutboxEvent{
|
|
{
|
|
ID: models.EventID{
|
|
EventIdempotencyKey: "account.saved:acc_123",
|
|
ConnectorID: connectorID,
|
|
},
|
|
EventType: events.EventTypeSavedAccounts,
|
|
EntityID: "acc_123",
|
|
Payload: json.RawMessage(`{"id":"acc_123","name":"Test Account 1"}`),
|
|
CreatedAt: time.Now().UTC(),
|
|
Status: models.OUTBOX_STATUS_PENDING,
|
|
ConnectorID: connectorID,
|
|
RetryCount: 0,
|
|
},
|
|
{
|
|
ID: models.EventID{
|
|
EventIdempotencyKey: "account.saved:acc_456",
|
|
ConnectorID: connectorID,
|
|
},
|
|
EventType: events.EventTypeSavedAccounts,
|
|
EntityID: "acc_456",
|
|
Payload: json.RawMessage(`{"id":"acc_456","name":"Test Account 2"}`),
|
|
CreatedAt: time.Now().UTC(),
|
|
Status: models.OUTBOX_STATUS_PENDING,
|
|
ConnectorID: connectorID,
|
|
RetryCount: 0,
|
|
},
|
|
}
|
|
|
|
// Poll events
|
|
s.EXPECT().
|
|
OutboxEventsPollPending(ctx, 100).
|
|
Return(testEvents, nil)
|
|
|
|
// First event publishes successfully
|
|
mockPublisher.EXPECT().
|
|
Publish(gomock.Any(), gomock.Any()).
|
|
Return(nil)
|
|
|
|
// Second event fails to publish
|
|
publishErr := errors.New("publish error")
|
|
mockPublisher.EXPECT().
|
|
Publish(gomock.Any(), gomock.Any()).
|
|
Return(publishErr)
|
|
|
|
// Mark second event as failed
|
|
s.EXPECT().
|
|
OutboxEventsMarkFailed(ctx, testEvents[1].ID, 1, publishErr).
|
|
Return(nil)
|
|
|
|
// Only first event should be batched for marking as processed
|
|
s.EXPECT().
|
|
OutboxEventsMarkProcessedAndRecordSent(ctx, gomock.Any(), gomock.Any()).
|
|
Do(func(_ context.Context, eventIDs []models.EventID, eventsSent []models.EventSent) {
|
|
Expect(eventIDs).To(HaveLen(1))
|
|
Expect(eventIDs[0]).To(Equal(testEvents[0].ID))
|
|
Expect(eventsSent).To(HaveLen(1))
|
|
Expect(eventsSent[0].ID).To(Equal(testEvents[0].ID))
|
|
}).
|
|
Return(nil)
|
|
|
|
err := act.OutboxPublishPendingEvents(ctx, 100)
|
|
Expect(err).To(BeNil())
|
|
})
|
|
})
|
|
})
|