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
92 lines
2.2 KiB
Go
92 lines
2.2 KiB
Go
package storage
|
|
|
|
import (
|
|
"context"
|
|
"crypto/rand"
|
|
"database/sql"
|
|
"os"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/formancehq/go-libs/v3/bun/bunconnect"
|
|
"github.com/formancehq/go-libs/v3/logging"
|
|
"github.com/formancehq/go-libs/v3/testing/docker"
|
|
"github.com/formancehq/go-libs/v3/testing/platform/pgtesting"
|
|
"github.com/formancehq/go-libs/v3/testing/utils"
|
|
"github.com/formancehq/payments/internal/models"
|
|
"github.com/stretchr/testify/require"
|
|
"github.com/uptrace/bun"
|
|
"github.com/uptrace/bun/dialect/pgdialect"
|
|
)
|
|
|
|
var (
|
|
srv *pgtesting.PostgresServer
|
|
bunDB *bun.DB
|
|
)
|
|
|
|
func TestMain(m *testing.M) {
|
|
utils.WithTestMain(func(t *utils.TestingTForMain) int {
|
|
srv = pgtesting.CreatePostgresServer(t, docker.NewPool(t, logging.Testing()))
|
|
|
|
db, err := sql.Open("pgx", srv.GetDSN())
|
|
if err != nil {
|
|
logging.Error(err)
|
|
os.Exit(1)
|
|
}
|
|
|
|
bunDB = bun.NewDB(db, pgdialect.New())
|
|
|
|
//return m.Run()
|
|
code := m.Run()
|
|
|
|
// Ensure the global bunDB is closed at the end of the test suite
|
|
_ = bunDB.Close()
|
|
return code
|
|
})
|
|
}
|
|
|
|
func newStore(t *testing.T) Storage {
|
|
t.Helper()
|
|
ctx := logging.TestingContext()
|
|
|
|
pgServer := srv.NewDatabase(t)
|
|
|
|
db, err := bunconnect.OpenSQLDB(ctx, pgServer.ConnectionOptions())
|
|
require.NoError(t, err)
|
|
|
|
key := make([]byte, 64)
|
|
_, err = rand.Read(key)
|
|
require.NoError(t, err)
|
|
|
|
err = Migrate(context.Background(), logging.Testing(), db, "test")
|
|
require.NoError(t, err)
|
|
|
|
//return newStorage(logging.Testing(), db, string(key))
|
|
|
|
st := newStorage(logging.Testing(), db, string(key))
|
|
|
|
// Ensure the store (and its DB connections) are closed before the per-test database is dropped.
|
|
t.Cleanup(func() {
|
|
_ = st.Close()
|
|
})
|
|
|
|
return st
|
|
}
|
|
|
|
func cleanupOutboxHelper(ctx context.Context, store Storage) func() {
|
|
return func() {
|
|
pendingEvents, _ := store.OutboxEventsPollPending(ctx, 1000)
|
|
var eventIDs []models.EventID
|
|
var eventsSent []models.EventSent
|
|
for _, event := range pendingEvents {
|
|
eventIDs = append(eventIDs, event.ID)
|
|
eventsSent = append(eventsSent, models.EventSent{})
|
|
}
|
|
if len(eventIDs) > 0 {
|
|
_ = store.OutboxEventsMarkProcessedAndRecordSent(ctx, eventIDs, eventsSent)
|
|
tomorrow := time.Now().AddDate(0, 0, 1)
|
|
_ = store.OutboxEventsDeleteOldProcessed(ctx, tomorrow)
|
|
}
|
|
}
|
|
}
|