Files
fgueryandGitHub a63e3a20a8 feat(EN-214): create events via storage instead of manually calling event activity (#594)
* 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
2025-11-24 16:05:42 +00:00

346 lines
10 KiB
Go

package storage
import (
"context"
"fmt"
"github.com/formancehq/go-libs/v3/bun/bunpaginate"
"github.com/formancehq/go-libs/v3/query"
"github.com/formancehq/go-libs/v3/time"
"github.com/formancehq/payments/internal/models"
"github.com/google/uuid"
"github.com/uptrace/bun"
)
type paymentServiceUser struct {
bun.BaseModel `bun:"payment_service_users"`
// Mandatory fields
ID uuid.UUID `bun:"id,pk,type:uuid,notnull"`
CreatedAt time.Time `bun:"created_at,type:timestamp without time zone,notnull"`
// Encrypted fields
Name string `bun:"decrypted_name,scanonly"`
StreetName *string `bun:"decrypted_street_name,scanonly"`
StreetNumber *string `bun:"decrypted_street_number,scanonly"`
City *string `bun:"decrypted_city,scanonly"`
PostalCode *string `bun:"decrypted_postal_code,scanonly"`
Region *string `bun:"decrypted_region,scanonly"`
Country *string `bun:"decrypted_country,scanonly"`
Email *string `bun:"decrypted_email,scanonly"`
PhoneNumber *string `bun:"decrypted_phone,scanonly"`
// Optional
Locale *string `bun:"locale,nullzero"`
// Optional fields with default
Metadata map[string]string `bun:"metadata,type:jsonb,nullzero,notnull,default:'{}'"`
// Relations
BankAccounts []bankAccount `bun:"rel:has-many,join:id=psu_id"`
}
func (s *store) PaymentServiceUsersCreate(ctx context.Context, psu models.PaymentServiceUser) error {
paymentServiceUser, relatedBankAccounts := fromPaymentServiceUserModels(psu)
tx, err := s.db.BeginTx(ctx, nil)
if err != nil {
return e("begin transaction", err)
}
var errTx error
defer func() {
rollbackOnTxError(ctx, &tx, errTx)
}()
_, err = tx.NewRaw(`
INSERT INTO payment_service_users (id, created_at, metadata, locale, name, street_name, street_number, city, region, postal_code, country, email, phone_number)
VALUES (?0, ?1, ?2, ?3,
pgp_sym_encrypt(?4::TEXT, ?13, ?14),
pgp_sym_encrypt(?5::TEXT, ?13, ?14),
pgp_sym_encrypt(?6::TEXT, ?13, ?14),
pgp_sym_encrypt(?7::TEXT, ?13, ?14),
pgp_sym_encrypt(?8::TEXT, ?13, ?14),
pgp_sym_encrypt(?9::TEXT, ?13, ?14),
pgp_sym_encrypt(?10::TEXT, ?13, ?14),
pgp_sym_encrypt(?11::TEXT, ?13, ?14),
pgp_sym_encrypt(?12::TEXT, ?13, ?14)
)
ON CONFLICT (id) DO NOTHING
RETURNING id
`, paymentServiceUser.ID, paymentServiceUser.CreatedAt, paymentServiceUser.Metadata,
paymentServiceUser.Locale, paymentServiceUser.Name, paymentServiceUser.StreetName, paymentServiceUser.StreetNumber, paymentServiceUser.City,
paymentServiceUser.Region, paymentServiceUser.PostalCode, paymentServiceUser.Country, paymentServiceUser.Email,
paymentServiceUser.PhoneNumber, s.configEncryptionKey, encryptionOptions,
).Exec(ctx)
if err != nil {
errTx = err
return e("insert psu", err)
}
if len(relatedBankAccounts) > 0 {
// Update related bank accounts
for _, bankAccountID := range relatedBankAccounts {
res, err := tx.NewUpdate().
Model((*bankAccount)(nil)).
Set("psu_id = ?", paymentServiceUser.ID).
Where("id = ?", bankAccountID).
Exec(ctx)
if err != nil {
errTx = err
return e("update bank account to add psu id", err)
}
rowsAffected, err := res.RowsAffected()
if err != nil {
errTx = err
return e("update bank account to add psu id", err)
}
if rowsAffected == 0 {
errTx = ErrNotFound
return e("bank account", ErrNotFound)
}
}
}
if err := tx.Commit(); err != nil {
errTx = err
return e("commit transaction", err)
}
return nil
}
func (s *store) PaymentServiceUsersGet(ctx context.Context, id uuid.UUID) (*models.PaymentServiceUser, error) {
var psu paymentServiceUser
query := s.db.NewSelect().
Model(&psu).
Column("id", "created_at", "metadata", "locale").
Where("id = ?", id).
Relation("BankAccounts")
query = s.paymentServiceUsersSelectDecryptColumnExpr(query)
err := query.
Scan(ctx)
if err != nil {
return nil, e("select psu", err)
}
res := toPaymentServiceUserModels(psu)
return &res, nil
}
func (s *store) PaymentServiceUsersDelete(ctx context.Context, paymentServiceUserID string) error {
res, err := s.db.NewDelete().
Model((*paymentServiceUser)(nil)).
Where("id = ?", paymentServiceUserID).
Exec(ctx)
if err != nil {
return e("delete psu", err)
}
rowsAffected, err := res.RowsAffected()
if err != nil {
return e("delete psu", err)
}
if rowsAffected == 0 {
return e("psu", ErrNotFound)
}
return nil
}
type PSUQuery struct{}
type ListPSUsQuery bunpaginate.OffsetPaginatedQuery[bunpaginate.PaginatedQueryOptions[PSUQuery]]
func NewListPSUQuery(opts bunpaginate.PaginatedQueryOptions[PSUQuery]) ListPSUsQuery {
return ListPSUsQuery{
Order: bunpaginate.OrderAsc,
PageSize: opts.PageSize,
Options: opts,
}
}
func (s *store) paymentServiceUsersQueryContext(qb query.Builder) (string, []any, error) {
return qb.Build(query.ContextFn(func(key, operator string, value any) (string, []any, error) {
switch {
case key == "id":
if operator != "$match" {
return "", nil, fmt.Errorf("'%s' column can only be used with $match: %w", key, ErrValidation)
}
return fmt.Sprintf("%s = ?", key), []any{value}, nil
case metadataRegex.Match([]byte(key)):
if operator != "$match" {
return "", nil, fmt.Errorf("'metadata' column can only be used with $match: %w", ErrValidation)
}
match := metadataRegex.FindAllStringSubmatch(key, 3)
key := "metadata"
return key + " @> ?", []any{map[string]any{
match[0][1]: value,
}}, nil
default:
return "", nil, fmt.Errorf("unknown key '%s' when building query: %w", key, ErrValidation)
}
}))
}
func (s *store) PaymentServiceUsersList(ctx context.Context, query ListPSUsQuery) (*bunpaginate.Cursor[models.PaymentServiceUser], error) {
var (
where string
args []any
err error
)
if query.Options.QueryBuilder != nil {
where, args, err = s.paymentServiceUsersQueryContext(query.Options.QueryBuilder)
if err != nil {
return nil, err
}
}
cursor, err := paginateWithOffset[bunpaginate.PaginatedQueryOptions[PSUQuery], paymentServiceUser](s, ctx,
(*bunpaginate.OffsetPaginatedQuery[bunpaginate.PaginatedQueryOptions[PSUQuery]])(&query),
func(query *bun.SelectQuery) *bun.SelectQuery {
query = query.Relation("BankAccounts")
query = query.Column("id", "created_at", "metadata", "locale")
query = s.paymentServiceUsersSelectDecryptColumnExpr(query)
if where != "" {
query = query.Where(where, args...)
}
query = query.Order("created_at DESC", "sort_id DESC")
return query
},
)
if err != nil {
return nil, e("failed to fetch accounts", err)
}
counterParties := make([]models.PaymentServiceUser, 0, len(cursor.Data))
for _, a := range cursor.Data {
counterParties = append(counterParties, toPaymentServiceUserModels(a))
}
return &bunpaginate.Cursor[models.PaymentServiceUser]{
PageSize: cursor.PageSize,
HasMore: cursor.HasMore,
Previous: cursor.Previous,
Next: cursor.Next,
Data: counterParties,
}, nil
}
func (s *store) PaymentServiceUsersAddBankAccount(ctx context.Context, psuID, bankAccountID uuid.UUID) error {
res, err := s.db.NewUpdate().
Model((*bankAccount)(nil)).
Set("psu_id = ?", psuID).
Where("id = ?", bankAccountID).
Exec(ctx)
if err != nil {
return e("update bank account to add psu id", err)
}
rowsAffected, err := res.RowsAffected()
if err != nil {
return e("update bank account to add psu id", err)
}
if rowsAffected == 0 {
return e("bank account", ErrNotFound)
}
return nil
}
func (s *store) paymentServiceUsersSelectDecryptColumnExpr(query *bun.SelectQuery) *bun.SelectQuery {
return query.
ColumnExpr("pgp_sym_decrypt(name, ?, ?) as decrypted_name", s.configEncryptionKey, encryptionOptions).
ColumnExpr("pgp_sym_decrypt(street_name, ?, ?) as decrypted_street_name", s.configEncryptionKey, encryptionOptions).
ColumnExpr("pgp_sym_decrypt(street_number, ?, ?) as decrypted_street_number", s.configEncryptionKey, encryptionOptions).
ColumnExpr("pgp_sym_decrypt(city, ?, ?) as decrypted_city", s.configEncryptionKey, encryptionOptions).
ColumnExpr("pgp_sym_decrypt(region, ?, ?) as decrypted_region", s.configEncryptionKey, encryptionOptions).
ColumnExpr("pgp_sym_decrypt(postal_code, ?, ?) as decrypted_postal_code", s.configEncryptionKey, encryptionOptions).
ColumnExpr("pgp_sym_decrypt(country, ?, ?) as decrypted_country", s.configEncryptionKey, encryptionOptions).
ColumnExpr("pgp_sym_decrypt(email, ?, ?) as decrypted_email", s.configEncryptionKey, encryptionOptions).
ColumnExpr("pgp_sym_decrypt(phone_number, ?, ?) as decrypted_phone", s.configEncryptionKey, encryptionOptions)
}
func fromPaymentServiceUserModels(from models.PaymentServiceUser) (paymentServiceUser, []uuid.UUID) {
psu := paymentServiceUser{
ID: from.ID,
CreatedAt: time.New(from.CreatedAt),
Name: from.Name,
Metadata: from.Metadata,
}
if from.Address != nil {
psu.StreetName = from.Address.StreetName
psu.StreetNumber = from.Address.StreetNumber
psu.City = from.Address.City
psu.PostalCode = from.Address.PostalCode
psu.Region = from.Address.Region
psu.Country = from.Address.Country
}
if from.ContactDetails != nil {
psu.Email = from.ContactDetails.Email
psu.PhoneNumber = from.ContactDetails.PhoneNumber
psu.Locale = from.ContactDetails.Locale
}
return psu, from.BankAccountIDs
}
func toPaymentServiceUserModels(from paymentServiceUser) models.PaymentServiceUser {
psu := models.PaymentServiceUser{
ID: from.ID,
CreatedAt: from.CreatedAt.Time,
Name: from.Name,
Metadata: from.Metadata,
}
psu.Address = fillAddress(from)
psu.ContactDetails = fillContactDetails(from)
psu.BankAccountIDs = make([]uuid.UUID, len(from.BankAccounts))
for i, bankAccount := range from.BankAccounts {
psu.BankAccountIDs[i] = bankAccount.ID
}
return psu
}
func fillAddress(from paymentServiceUser) *models.Address {
if from.StreetName == nil && from.StreetNumber == nil && from.City == nil && from.PostalCode == nil && from.Region == nil && from.Country == nil {
return nil
}
return &models.Address{
StreetName: from.StreetName,
StreetNumber: from.StreetNumber,
City: from.City,
PostalCode: from.PostalCode,
Region: from.Region,
Country: from.Country,
}
}
func fillContactDetails(from paymentServiceUser) *models.ContactDetails {
if from.Email == nil && from.PhoneNumber == nil && from.Locale == nil {
return nil
}
return &models.ContactDetails{
Email: from.Email,
PhoneNumber: from.PhoneNumber,
Locale: from.Locale,
}
}