payments/internal/storage/payments.go

package storage

import (
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	"math/big"
	"time"

	"github.com/hanzo-fi/go-libs/v5/pkg/query"
	"github.com/hanzo-fi/go-libs/v5/pkg/storage/bun/paginate"
	internalTime "github.com/hanzo-fi/go-libs/v5/pkg/types/time"
	internalEvents "github.com/hanzo-fi/payments/internal/events"
	"github.com/hanzo-fi/payments/pkg/domain/models"
	"github.com/hanzo-fi/payments/pkg/events"
	"github.com/google/uuid"
	"github.com/pkg/errors"
	"github.com/uptrace/bun"
)

type payment struct {
	bun.BaseModel `bun:"table:payments"`

	// Mandatory fields
	ID            models.PaymentID     `bun:"id,pk,type:character varying,notnull"`
	ConnectorID   models.ConnectorID   `bun:"connector_id,type:character varying,notnull"`
	Reference     string               `bun:"reference,type:text,notnull"`
	CreatedAt     internalTime.Time    `bun:"created_at,type:timestamp without time zone,notnull"`
	Type          models.PaymentType   `bun:"type,type:text,notnull"`
	InitialAmount *big.Int             `bun:"initial_amount,type:numeric,notnull"`
	Amount        *big.Int             `bun:"amount,type:numeric,notnull"`
	Asset         string               `bun:"asset,type:text,notnull"`
	Scheme        models.PaymentScheme `bun:"scheme,type:text,notnull"`

	// Scan only fields
	Status models.PaymentStatus `bun:"status,type:text,notnull,scanonly"`

	// Optional fields
	// c.f.: https://bun.uptrace.dev/guide/models.html#nulls
	SourceAccountID         *models.AccountID `bun:"source_account_id,type:character varying,nullzero"`
	DestinationAccountID    *models.AccountID `bun:"destination_account_id,type:character varying,nullzero"`
	PsuID                   *uuid.UUID        `bun:"psu_id,type:uuid,nullzero"`
	OpenBankingConnectionID *string           `bun:"open_banking_connection_id,type:character varying,nullzero"`

	// Optional fields with default
	// c.f. https://bun.uptrace.dev/guide/models.html#default
	Metadata map[string]string `bun:"metadata,type:jsonb,nullzero,notnull,default:'{}'"`
}

type paymentAdjustment struct {
	bun.BaseModel `bun:"table:payment_adjustments"`

	// Mandatory fields
	ID        models.PaymentAdjustmentID `bun:"id,pk,type:character varying,notnull"`
	PaymentID models.PaymentID           `bun:"payment_id,type:character varying,notnull"`
	Reference string                     `bun:"reference,type:text,notnull"`
	CreatedAt internalTime.Time          `bun:"created_at,type:timestamp without time zone,notnull"`
	Status    models.PaymentStatus       `bun:"status,type:text,notnull"`
	Raw       json.RawMessage            `bun:"raw,type:json,notnull"`

	// Optional fields
	// c.f.: https://bun.uptrace.dev/guide/models.html#nulls
	Amount *big.Int `bun:"amount,type:numeric,nullzero"`
	Asset  *string  `bun:"asset,type:text,nullzero"`

	// Optional fields with default
	// c.f. https://bun.uptrace.dev/guide/models.html#default
	Metadata map[string]string `bun:"metadata,type:jsonb,nullzero,notnull,default:'{}'"`
}

func (s *store) PaymentsUpsert(ctx context.Context, payments []models.Payment) error {
	paymentsToInsert := make([]payment, 0, len(payments))
	adjustmentsToInsert := make([]paymentAdjustment, 0)
	for _, p := range payments {
		paymentsToInsert = append(paymentsToInsert, fromPaymentModels(p))
		for _, a := range p.Adjustments {
			adjustmentsToInsert = append(adjustmentsToInsert, fromPaymentAdjustmentModels(a))
		}
	}

	tx, err := s.db.BeginTx(ctx, &sql.TxOptions{})
	if err != nil {
		return e("failed to create transaction", err)
	}
	defer func() {
		rollbackOnTxError(ctx, &tx, err)
	}()

	if len(paymentsToInsert) > 0 {
		_, err = tx.NewInsert().
			Model(&paymentsToInsert).
			On("CONFLICT (id) DO NOTHING").
			Exec(ctx)
		if err != nil {
			return e("failed to insert payments", err)
		}
	}

	// Insert adjustments before applying any amount delta. Temporal activities are
	// at-least-once, so a replayed batch must be idempotent: the RETURNING clause tells
	// us which adjustments were actually inserted, and only those may move a payment's
	// amount. Re-applying deltas for already-stored adjustments would, for example,
	// subtract a refund twice (EN-1085 / C1).
	var insertedAdjustments []paymentAdjustment
	if len(adjustmentsToInsert) > 0 {
		err = tx.NewInsert().
			Model(&adjustmentsToInsert).
			On("CONFLICT (id) DO NOTHING").
			Returning("*").
			Scan(ctx, &insertedAdjustments)
		if err != nil {
			return e("failed to insert adjustments", err)
		}
	}

	// paymentMap resolves the full payment model for an inserted adjustment, both for the
	// amount deltas below and for the outbox events further down.
	paymentMap := make(map[models.PaymentID]models.Payment, len(payments))
	for _, p := range payments {
		paymentMap[p.ID] = p
	}

	// Refund/capture amount deltas are driven from the adjustments that were actually
	// inserted (the RETURNING set, already deduplicated by primary key). Applying one delta
	// per inserted row makes both a replayed batch and a batch that repeats the same
	// adjustment idempotent: a refund cannot be subtracted twice (EN-1085 / C1). Each delta
	// is accumulated into a freshly-allocated big.Int so we never mutate a caller's or a
	// persisted adjustment's amount in place (EN-1085 / C2).
	paymentsRefundedSeen := make(map[models.PaymentID]int)
	paymentsRefunded := make([]payment, 0)
	paymentsCapturedSeen := make(map[models.PaymentID]int)
	paymentsCaptured := make([]payment, 0)
	for _, adj := range insertedAdjustments {
		// Amount is optional on adjustments; without it there is no delta to apply.
		if adj.Amount == nil {
			continue
		}
		p, ok := paymentMap[adj.PaymentID]
		if !ok {
			continue
		}
		switch adj.Status {
		case models.PAYMENT_STATUS_REFUNDED:
			if i, ok := paymentsRefundedSeen[p.ID]; ok {
				paymentsRefunded[i].Amount.Add(paymentsRefunded[i].Amount, adj.Amount)
			} else {
				res := fromPaymentModels(p)
				res.Amount = new(big.Int).Set(adj.Amount)
				paymentsRefunded = append(paymentsRefunded, res)
				paymentsRefundedSeen[p.ID] = len(paymentsRefunded) - 1
			}
		case models.PAYMENT_STATUS_CAPTURE, models.PAYMENT_STATUS_REFUND_REVERSED:
			if i, ok := paymentsCapturedSeen[p.ID]; ok {
				paymentsCaptured[i].Amount.Add(paymentsCaptured[i].Amount, adj.Amount)
			} else {
				res := fromPaymentModels(p)
				res.Amount = new(big.Int).Set(adj.Amount)
				paymentsCaptured = append(paymentsCaptured, res)
				paymentsCapturedSeen[p.ID] = len(paymentsCaptured) - 1
			}
		}
	}

	// initial_amount is set absolutely (= EXCLUDED.initial_amount), not as a delta, so it
	// is already idempotent on replay. Preserve the original "last amount adjustment in the
	// input wins" semantics by walking the input in order; copy the amount to avoid aliasing
	// (EN-1085 / C2).
	paymentsInitialAmountToAdjustSeen := make(map[models.PaymentID]int)
	paymentsInitialAmountToAdjust := make([]payment, 0)
	for _, p := range payments {
		for _, a := range p.Adjustments {
			if a.Status != models.PAYMENT_STATUS_AMOUNT_ADJUSTMENT || a.Amount == nil {
				continue
			}
			if i, ok := paymentsInitialAmountToAdjustSeen[p.ID]; ok {
				paymentsInitialAmountToAdjust[i].InitialAmount = new(big.Int).Set(a.Amount)
			} else {
				res := fromPaymentModels(p)
				res.InitialAmount = new(big.Int).Set(a.Amount)
				paymentsInitialAmountToAdjust = append(paymentsInitialAmountToAdjust, res)
				paymentsInitialAmountToAdjustSeen[p.ID] = len(paymentsInitialAmountToAdjust) - 1
			}
		}
	}

	if len(paymentsInitialAmountToAdjust) > 0 {
		_, err = tx.NewInsert().
			Model(&paymentsInitialAmountToAdjust).
			On("CONFLICT (id) DO UPDATE").
			Set("initial_amount = EXCLUDED.initial_amount").
			Exec(ctx)
		if err != nil {
			return e("failed to update payment", err)
		}
	}

	if len(paymentsCaptured) > 0 {
		_, err = tx.NewInsert().
			Model(&paymentsCaptured).
			On("CONFLICT (id) DO UPDATE").
			Set("amount = payment.amount + EXCLUDED.amount").
			Exec(ctx)
		if err != nil {
			return e("failed to update payment", err)
		}
	}

	if len(paymentsRefunded) > 0 {
		_, err = tx.NewInsert().
			Model(&paymentsRefunded).
			On("CONFLICT (id) DO UPDATE").
			Set("amount = payment.amount - EXCLUDED.amount").
			Exec(ctx)
		if err != nil {
			return e("failed to update payment", err)
		}
	}

	// Create outbox events for each inserted adjustment
	if len(insertedAdjustments) > 0 {
		outboxEvents := make([]models.OutboxEvent, 0, len(insertedAdjustments))
		for _, adj := range insertedAdjustments {
			payment, ok := paymentMap[adj.PaymentID]
			if !ok {
				// This shouldn't happen, but skip if payment not found
				continue
			}

			// Create the event payload matching EventsSendPayment format
			payload := internalEvents.PaymentMessagePayload{
				ID:            payment.ID.String(),
				Reference:     payment.Reference,
				Type:          payment.Type.String(),
				Status:        adj.Status.String(),
				InitialAmount: payment.InitialAmount,
				Amount:        payment.Amount,
				Scheme:        payment.Scheme.String(),
				Asset:         payment.Asset,
				CreatedAt:     payment.CreatedAt,
				ConnectorID:   payment.ConnectorID.String(),
				Provider:      models.ToV3Provider(payment.ConnectorID.Provider),
				RawData:       adj.Raw,
				Metadata:      payment.Metadata,
			}

			if payment.SourceAccountID != nil {
				payload.SourceAccountID = payment.SourceAccountID.String()
			}

			if payment.DestinationAccountID != nil {
				payload.DestinationAccountID = payment.DestinationAccountID.String()
			}

			var payloadBytes []byte
			payloadBytes, err = json.Marshal(&payload)
			if err != nil {
				return e("failed to marshal payment event payload", err)
			}

			// Convert adjustment back to model to get idempotency key
			adjustmentModel := toPaymentAdjustmentModels(adj)
			connectorID := payment.ConnectorID
			outboxEvent := models.OutboxEvent{
				ID: models.EventID{
					EventIdempotencyKey: adjustmentModel.IdempotencyKey(),
					ConnectorID:         &connectorID,
				},
				EventType:   events.EventTypeSavedPayments,
				EntityID:    payment.ID.String(),
				Payload:     payloadBytes,
				CreatedAt:   time.Now().UTC(),
				Status:      models.OUTBOX_STATUS_PENDING,
				ConnectorID: &connectorID,
			}

			outboxEvents = append(outboxEvents, outboxEvent)
		}

		// Insert outbox events in the same transaction
		if len(outboxEvents) > 0 {
			if err = s.OutboxEventsInsert(ctx, tx, outboxEvents); err != nil {
				return err
			}
		}
	}

	if err = tx.Commit(); err != nil {
		return e("failed to commit transaction", err)
	}
	return nil
}

func (s *store) PaymentsUpdateMetadata(ctx context.Context, id models.PaymentID, metadata map[string]string) error {
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return e("update payment metadata", err)
	}
	defer func() {
		rollbackOnTxError(ctx, &tx, err)
	}()

	var payment payment
	err = tx.NewSelect().
		Model(&payment).
		Column("id", "metadata").
		Where("id = ?", id).
		Scan(ctx)
	if err != nil {
		return e("update payment metadata", err)
	}

	if payment.Metadata == nil {
		payment.Metadata = make(map[string]string)
	}

	for k, v := range metadata {
		payment.Metadata[k] = v
	}

	_, err = tx.NewUpdate().
		Model(&payment).
		Column("metadata").
		Where("id = ?", id).
		Exec(ctx)
	if err != nil {
		return e("update payment metadata", err)
	}

	if err = tx.Commit(); err != nil {
		return e("failed to commit transaction", err)
	}
	return nil
}

func (s *store) PaymentsGet(ctx context.Context, id models.PaymentID) (*models.Payment, error) {
	var payment payment

	err := s.db.NewSelect().
		Model(&payment).
		Where("id = ?", id).
		Scan(ctx)
	if err != nil {
		return nil, e("failed to get payment", err)
	}

	var ajs []paymentAdjustment
	err = s.db.NewSelect().
		Model(&ajs).
		Where("payment_id = ?", id).
		Order("created_at DESC", "sort_id DESC").
		Scan(ctx)
	if err != nil {
		return nil, e("failed to get payment adjustments", err)
	}

	adjustments := make([]models.PaymentAdjustment, 0, len(ajs))
	for _, a := range ajs {
		adjustments = append(adjustments, toPaymentAdjustmentModels(a))
	}

	status := models.PAYMENT_STATUS_PENDING
	if len(adjustments) > 0 {
		// This list is ordered by created_at DESC, so the first element is the
		// last adjustment, and we want the last status.
		status = adjustments[0].Status
	}
	res := toPaymentModels(payment, status)
	res.Adjustments = adjustments
	return &res, nil
}

func (s *store) PaymentsGetByReference(ctx context.Context, reference string, connectorID models.ConnectorID) (*models.Payment, error) {
	var payment payment

	err := s.db.NewSelect().
		Model(&payment).
		Where("reference = ?", reference).
		Where("connector_id = ?", connectorID).
		Scan(ctx)
	if err != nil {
		return nil, e("failed to get payment", err)
	}

	var ajs []paymentAdjustment
	err = s.db.NewSelect().
		Model(&ajs).
		Where("payment_id = ?", payment.ID).
		Order("created_at DESC", "sort_id DESC").
		Scan(ctx)
	if err != nil {
		return nil, e("failed to get payment adjustments", err)
	}

	adjustments := make([]models.PaymentAdjustment, 0, len(ajs))
	for _, a := range ajs {
		adjustments = append(adjustments, toPaymentAdjustmentModels(a))
	}

	status := models.PAYMENT_STATUS_PENDING
	if len(adjustments) > 0 {
		// This list is ordered by created_at DESC, so the first element is the
		// last adjustment, and we want the last status.
		status = adjustments[0].Status
	}
	res := toPaymentModels(payment, status)
	res.Adjustments = adjustments
	return &res, nil
}

func (s *store) PaymentsDeleteFromConnectorID(ctx context.Context, connectorID models.ConnectorID) error {
	_, err := s.db.NewDelete().
		Model((*payment)(nil)).
		Where("connector_id = ?", connectorID).
		Exec(ctx)

	return e("failed to delete payments", err)
}

// PaymentsDeleteFromConnectorIDBatch deletes a batch of payments for a given connector ID
// and returns the number of rows affected
func (s *store) PaymentsDeleteFromConnectorIDBatch(ctx context.Context, connectorID models.ConnectorID, batchSize int) (int, error) {
	if batchSize <= 0 {
		return 0, fmt.Errorf("invalid batchSize %d for connectorID %s: %w", batchSize, connectorID.String(), ErrValidation)
	}

	result, err := s.db.NewDelete().
		Model((*payment)(nil)).
		Where("id IN (SELECT id FROM payments WHERE connector_id = ? LIMIT ?)", connectorID, batchSize).
		Exec(ctx)
	if err != nil {
		return 0, e("failed to delete payments batch", err)
	}

	rowsAffected, err := result.RowsAffected()
	if err != nil {
		return 0, e("failed to get rows affected", err)
	}

	return int(rowsAffected), nil
}

func (s *store) PaymentsDelete(ctx context.Context, id models.PaymentID) error {
	_, err := s.db.NewDelete().
		Model((*payment)(nil)).
		Where("id = ?", id).
		Exec(ctx)

	return e("failed to delete payment", err)
}

// PaymentsDeleteFromReference TODO this deletion method is the only one emitting outbox events.
// Using the outbox pattern makes this obvious, but others flows did not either before that pattern was set up.
func (s *store) PaymentsDeleteFromReference(ctx context.Context, reference string, connectorID models.ConnectorID) error {
	tx, err := s.db.BeginTx(ctx, &sql.TxOptions{})
	if err != nil {
		return e("failed to create transaction", err)
	}
	defer func() {
		rollbackOnTxError(ctx, &tx, err)
	}()

	// Get the payment before deleting it (to create outbox event)
	var p payment
	err = tx.NewSelect().
		Model(&p).
		Where("reference = ?", reference).
		Where("connector_id = ?", connectorID).
		Scan(ctx)
	if err != nil {
		pErr := e("failed to get payment", err)
		if errors.Is(pErr, ErrNotFound) {
			// Payment doesn't exist, nothing to delete or create event for
			if commitErr := tx.Commit(); commitErr != nil {
				err = commitErr
				return e("failed to commit transaction", commitErr)
			}
			err = nil // CLean up the error as we already commited the transaction, no rollback required
			return nil
		}
		return pErr
	}

	// Delete the payment
	_, err = tx.NewDelete().
		Model((*payment)(nil)).
		Where("reference = ?", reference).
		Where("connector_id = ?", connectorID).
		Exec(ctx)
	if err != nil {
		return e("failed to delete payment", err)
	}

	// Create outbox event for deleted payment
	paymentID := p.ID
	payload := internalEvents.PaymentDeletedMessagePayload{
		ID: paymentID.String(),
	}

	var payloadBytes []byte
	payloadBytes, err = json.Marshal(payload)
	if err != nil {
		return e("failed to marshal payment deleted event payload", err)
	}

	outboxEvent := models.OutboxEvent{
		ID: models.EventID{
			EventIdempotencyKey: fmt.Sprintf("delete:%s", paymentID.String()),
			ConnectorID:         &connectorID,
		},
		EventType:   events.EventTypeDeletedPayments,
		EntityID:    paymentID.String(),
		Payload:     payloadBytes,
		CreatedAt:   time.Now().UTC(),
		Status:      models.OUTBOX_STATUS_PENDING,
		ConnectorID: &connectorID,
	}

	if err = s.OutboxEventsInsert(ctx, tx, []models.OutboxEvent{outboxEvent}); err != nil {
		return err
	}

	if err = tx.Commit(); err != nil {
		return e("failed to commit transaction", err)
	}
	return nil
}

func (s *store) PaymentsDeleteFromAccountID(ctx context.Context, accountID models.AccountID) error {
	_, err := s.db.NewDelete().
		Model((*payment)(nil)).
		Where("source_account_id = ? OR destination_account_id = ?", accountID, accountID).
		Exec(ctx)

	return e("failed to delete payments", err)
}

func (s *store) PaymentsDeleteFromPSUID(ctx context.Context, psuID uuid.UUID) error {
	_, err := s.db.NewDelete().
		Model((*payment)(nil)).
		Where("psu_id = ?", psuID).
		Exec(ctx)

	return e("failed to delete payments", err)
}

func (s *store) PaymentsDeleteFromConnectorIDAndPSUID(ctx context.Context, connectorID models.ConnectorID, psuID uuid.UUID) error {
	_, err := s.db.NewDelete().
		Model((*payment)(nil)).
		Where("connector_id = ?", connectorID).
		Where("psu_id = ?", psuID).
		Exec(ctx)

	return e("failed to delete payments", err)
}

func (s *store) PaymentsDeleteFromOpenBankingConnectionID(ctx context.Context, psuID uuid.UUID, connectorID models.ConnectorID, openBankingConnectionID string) error {
	_, err := s.db.NewDelete().
		Model((*payment)(nil)).
		Where("psu_id = ?", psuID).
		Where("connector_id = ?", connectorID).
		Where("open_banking_connection_id = ?", openBankingConnectionID).
		Exec(ctx)

	return e("failed to delete payments", err)
}

type PaymentQuery struct{}

type ListPaymentsQuery paginate.OffsetPaginatedQuery[paginate.PaginatedQueryOptions[PaymentQuery]]

func NewListPaymentsQuery(opts paginate.PaginatedQueryOptions[PaymentQuery]) ListPaymentsQuery {
	return ListPaymentsQuery{
		PageSize: opts.PageSize,
		Order:    paginate.OrderAsc,
		Options:  opts,
	}
}

func (s *store) paymentsQueryContext(qb query.Builder) (string, []any, error) {
	where, args, err := qb.Build(query.ContextFn(func(key, operator string, value any) (string, []any, error) {
		switch {
		case key == "reference",
			key == "id",
			key == "connector_id",
			key == "type",
			key == "asset",
			key == "scheme",
			key == "status",
			key == "source_account_id",
			key == "destination_account_id",
			key == "psu_id",
			key == "open_banking_connection_id":
			if operator != "$match" {
				return "", nil, e(fmt.Sprintf("'%s' column can only be used with $match", key), ErrValidation)
			}
			return fmt.Sprintf("%s = ?", key), []any{value}, nil

		case key == "initial_amount",
			key == "amount":
			return fmt.Sprintf("%s %s ?", key, query.DefaultComparisonOperatorsMapping[operator]), []any{value}, nil
		case metadataRegex.Match([]byte(key)):
			if operator != "$match" {
				return "", nil, e("'metadata' column can only be used with $match", 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)
		}
	}))

	return where, args, err
}

func (s *store) PaymentsList(ctx context.Context, q ListPaymentsQuery) (*paginate.Cursor[models.Payment], error) {
	var (
		where string
		args  []any
		err   error
	)
	if q.Options.QueryBuilder != nil {
		where, args, err = s.paymentsQueryContext(q.Options.QueryBuilder)
		if err != nil {
			return nil, err
		}
	}

	// TODO(polo): should fetch the adjustments and get the last status and amount?
	cursor, err := paginateWithOffset[paginate.PaginatedQueryOptions[PaymentQuery], payment](s, ctx,
		(*paginate.OffsetPaginatedQuery[paginate.PaginatedQueryOptions[PaymentQuery]])(&q),
		func(query *bun.SelectQuery) *bun.SelectQuery {
			if where != "" {
				query = query.Where(where, args...)
			}

			query.Column("payment.*", "apd.status").
				Join(`join lateral (
				select status
				from payment_adjustments apd
				where payment_id = payment.id
				order by created_at desc, sort_id desc
				limit 1
			) apd on true`)

			// TODO(polo): sorter ?
			query = query.Order("created_at DESC", "sort_id DESC")

			return query
		},
	)
	if err != nil {
		return nil, e("failed to fetch payments", err)
	}

	payments := make([]models.Payment, 0, len(cursor.Data))
	for _, p := range cursor.Data {
		payments = append(payments, toPaymentModels(p, p.Status))
	}

	return &paginate.Cursor[models.Payment]{
		PageSize: cursor.PageSize,
		HasMore:  cursor.HasMore,
		Previous: cursor.Previous,
		Next:     cursor.Next,
		Data:     payments,
	}, nil
}

func fromPaymentModels(from models.Payment) payment {
	return payment{
		ID:                      from.ID,
		ConnectorID:             from.ConnectorID,
		Reference:               from.Reference,
		CreatedAt:               internalTime.New(from.CreatedAt),
		Type:                    from.Type,
		InitialAmount:           from.InitialAmount,
		Amount:                  from.Amount,
		Asset:                   from.Asset,
		Scheme:                  from.Scheme,
		SourceAccountID:         from.SourceAccountID,
		DestinationAccountID:    from.DestinationAccountID,
		PsuID:                   from.PsuID,
		OpenBankingConnectionID: from.OpenBankingConnectionID,
		Metadata:                from.Metadata,
	}
}

func toPaymentModels(payment payment, status models.PaymentStatus) models.Payment {
	return models.Payment{
		ID:                      payment.ID,
		ConnectorID:             payment.ConnectorID,
		InitialAmount:           payment.InitialAmount,
		Reference:               payment.Reference,
		CreatedAt:               payment.CreatedAt.Time,
		Type:                    payment.Type,
		Amount:                  payment.Amount,
		Asset:                   payment.Asset,
		Scheme:                  payment.Scheme,
		Status:                  status,
		SourceAccountID:         payment.SourceAccountID,
		DestinationAccountID:    payment.DestinationAccountID,
		PsuID:                   payment.PsuID,
		OpenBankingConnectionID: payment.OpenBankingConnectionID,
		Metadata:                payment.Metadata,
	}
}

func fromPaymentAdjustmentModels(from models.PaymentAdjustment) paymentAdjustment {
	// Copy the amount so the persisted row never aliases the caller's big.Int, which
	// could otherwise be mutated in place while accumulating deltas (EN-1085 / C2).
	var amount *big.Int
	if from.Amount != nil {
		amount = new(big.Int).Set(from.Amount)
	}
	return paymentAdjustment{
		ID:        from.ID,
		PaymentID: from.ID.PaymentID,
		Reference: from.Reference,
		CreatedAt: internalTime.New(from.CreatedAt),
		Status:    from.Status,
		Amount:    amount,
		Asset:     from.Asset,
		Metadata:  from.Metadata,
		Raw:       from.Raw,
	}
}

func toPaymentAdjustmentModels(from paymentAdjustment) models.PaymentAdjustment {
	return models.PaymentAdjustment{
		ID:        from.ID,
		Reference: from.Reference,
		CreatedAt: from.CreatedAt.Time,
		Status:    from.Status,
		Amount:    from.Amount,
		Asset:     from.Asset,
		Metadata:  from.Metadata,
		Raw:       from.Raw,
	}
}