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
199 lines
5.6 KiB
Go
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,
|
|
}
|
|
}
|