Files
payments/internal/events/payment_initiation.go
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

199 lines
5.6 KiB
Go

package events
import (
"encoding/json"
"fmt"
"math/big"
"time"
"github.com/formancehq/go-libs/v3/pointer"
"github.com/formancehq/go-libs/v3/publish"
"github.com/formancehq/payments/internal/models"
"github.com/formancehq/payments/pkg/events"
)
type PaymentInitiationMessagePayload struct {
// Mandatory fields
ID string `json:"id"`
ConnectorID string `json:"connectorID"`
Provider string `json:"provider"`
Reference string `json:"reference"`
CreatedAt time.Time `json:"createdAt"`
ScheduledAt time.Time `json:"scheduledAt"`
Description string `json:"description"`
Type string `json:"type"`
Amount *big.Int `json:"amount"`
Asset string `json:"asset"`
// Optional fields
SourceAccountID string `json:"sourceAccountID,omitempty"`
DestinationAccountID string `json:"destinationAccountID,omitempty"`
Metadata map[string]string `json:"metadata,omitempty"`
}
func (p *PaymentInitiationMessagePayload) MarshalJSON() ([]byte, error) {
type Alias PaymentInitiationMessagePayload
var amountStr *string
if p.Amount != nil {
s := p.Amount.String()
amountStr = &s
}
return json.Marshal(&struct {
Amount *string `json:"amount"`
*Alias
}{
Amount: amountStr,
Alias: (*Alias)(p),
})
}
func (p *PaymentInitiationMessagePayload) UnmarshalJSON(data []byte) error {
type Alias PaymentInitiationMessagePayload
aux := &struct {
Amount *string `json:"amount"`
*Alias
}{
Alias: (*Alias)(p),
}
if err := json.Unmarshal(data, &aux); err != nil {
return err
}
if aux.Amount != nil {
bi := new(big.Int)
if _, ok := bi.SetString(*aux.Amount, 10); !ok {
return fmt.Errorf("invalid amount string: %s", *aux.Amount)
}
p.Amount = bi
} else {
p.Amount = nil
}
return nil
}
type PaymentInitiationAdjustmentMessagePayload struct {
// Mandatory fields
ID string `json:"id"`
PaymentInitiationID string `json:"paymentInitiationID"`
Status string `json:"status"`
// Optional fields
Amount *big.Int `json:"amount,omitempty"`
Asset *string `json:"asset,omitempty"`
Error *string `json:"error,omitempty"`
Metadata map[string]string `json:"metadata,omitempty"`
}
func (p *PaymentInitiationAdjustmentMessagePayload) MarshalJSON() ([]byte, error) {
type Alias PaymentInitiationAdjustmentMessagePayload
var amountStr *string
if p.Amount != nil {
s := p.Amount.String()
amountStr = &s
}
return json.Marshal(&struct {
Amount *string `json:"amount,omitempty"`
*Alias
}{
Amount: amountStr,
Alias: (*Alias)(p),
})
}
func (p *PaymentInitiationAdjustmentMessagePayload) UnmarshalJSON(data []byte) error {
type Alias PaymentInitiationAdjustmentMessagePayload
aux := &struct {
Amount *string `json:"amount,omitempty"`
*Alias
}{
Alias: (*Alias)(p),
}
if err := json.Unmarshal(data, &aux); err != nil {
return err
}
if aux.Amount != nil {
bi := new(big.Int)
if _, ok := bi.SetString(*aux.Amount, 10); !ok {
return fmt.Errorf("invalid amount string: %s", *aux.Amount)
}
p.Amount = bi
} else {
p.Amount = nil
}
return nil
}
type PaymentInitiationRelatedPaymentMessagePayload struct {
PaymentInitiationID string `json:"paymentInitiationID"`
PaymentID string `json:"paymentID"`
}
func (e Events) NewEventSavedPaymentInitiation(pi models.PaymentInitiation) publish.EventMessage {
payload := PaymentInitiationMessagePayload{
ID: pi.ID.String(),
ConnectorID: pi.ConnectorID.String(),
Provider: pi.ConnectorID.Provider,
Reference: pi.Reference,
CreatedAt: pi.CreatedAt,
ScheduledAt: pi.ScheduledAt,
Description: pi.Description,
Type: pi.Type.String(),
SourceAccountID: pi.SourceAccountID.String(),
DestinationAccountID: pi.DestinationAccountID.String(),
Amount: pi.Amount,
Asset: pi.Asset,
Metadata: pi.Metadata,
}
return publish.EventMessage{
IdempotencyKey: pi.IdempotencyKey(),
Date: time.Now().UTC(),
App: events.EventApp,
Version: events.EventVersion,
Type: events.EventTypeSavedPaymentInitiation,
Payload: payload,
}
}
func (e Events) NewEventSavedPaymentInitiationAdjustment(adj models.PaymentInitiationAdjustment) publish.EventMessage {
payload := PaymentInitiationAdjustmentMessagePayload{
ID: adj.ID.String(),
PaymentInitiationID: adj.ID.PaymentInitiationID.String(),
Status: adj.Status.String(),
Amount: adj.Amount,
Asset: adj.Asset,
Error: func() *string {
if adj.Error == nil {
return nil
}
return pointer.For(adj.Error.Error())
}(),
Metadata: adj.Metadata,
}
return publish.EventMessage{
IdempotencyKey: adj.IdempotencyKey(),
Date: time.Now().UTC(),
App: events.EventApp,
Version: events.EventVersion,
Type: events.EventTypeSavedPaymentInitiationAdjustment,
Payload: payload,
}
}
func (e Events) NewEventSavedPaymentInitiationRelatedPayment(relatedPayment models.PaymentInitiationRelatedPayments) publish.EventMessage {
payload := PaymentInitiationRelatedPaymentMessagePayload{
PaymentInitiationID: relatedPayment.PaymentInitiationID.String(),
PaymentID: relatedPayment.PaymentID.String(),
}
return publish.EventMessage{
IdempotencyKey: relatedPayment.IdempotencyKey(),
Date: time.Now().UTC(),
App: events.EventApp,
Version: events.EventVersion,
Type: events.EventTypeSavedPaymentInitiationRelatedPayment,
Payload: payload,
}
}