mirror of
https://github.com/hanzo-fi/payments.git
synced 2026-10-30 20:31:20 +00:00
* 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
346 lines
10 KiB
Go
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,
|
|
}
|
|
}
|