প্রজেক্ট: পেমেন্ট লেজার

package ledger

import (
	"context"
	"crypto/sha256"
	"database/sql"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"sort"
	"strings"

	"github.com/google/uuid"
)

// --- Domain types ---

type EntryInput struct {
	AccountRef string `json:"accountRef"`
	// Signed minor units. Debit positive, credit negative.
	Amount int64 `json:"amount"`
}

type PostRequest struct {
	IdempotencyKey string                 `json:"idempotencyKey"`
	Kind           string                 `json:"kind"`
	Currency       string                 `json:"currency"`
	Description    string                 `json:"description"`
	Entries        []EntryInput           `json:"entries"`
	Metadata       map[string]interface{} `json:"metadata"`
	ReversesRef    *string                `json:"reversesRef,omitempty"`
}

type PostResult struct {
	TransactionRef string            `json:"transactionRef"`
	SequenceNo     int64             `json:"sequenceNo"`
	Replayed       bool              `json:"replayed"`
	Balances       map[string]int64  `json:"balances"`
}

type LedgerError struct {
	Code    string
	Message string
}

func (e *LedgerError) Error() string { return e.Code + ": " + e.Message }

func newErr(code, format string, args ...interface{}) error {
	return &LedgerError{Code: code, Message: fmt.Sprintf(format, args...)}
}

// --- Helpers ---

func canonicalHash(req PostRequest) string {
	type ce struct {
		A string `json:"a"`
		V int64  `json:"v"`
	}
	entries := make([]ce, 0, len(req.Entries))
	for _, e := range req.Entries {
		entries = append(entries, ce{A: e.AccountRef, V: e.Amount})
	}
	sort.Slice(entries, func(i, j int) bool {
		if entries[i].A != entries[j].A {
			return entries[i].A < entries[j].A
		}
		return entries[i].V < entries[j].V
	})
	rev := ""
	if req.ReversesRef != nil {
		rev = *req.ReversesRef
	}
	payload, _ := json.Marshal(map[string]interface{}{
		"kind":        req.Kind,
		"currency":    req.Currency,
		"reversesRef": rev,
		"entries":     entries,
	})
	sum := sha256.Sum256(payload)
	return hex.EncodeToString(sum[:])
}

func assertBalanced(entries []EntryInput) error {
	if len(entries) < 2 {
		return newErr("UNBALANCED", "a transaction needs at least two entries")
	}
	var sum int64
	for _, e := range entries {
		if e.Amount == 0 {
			return newErr("ZERO_ENTRY", "zero amount entry for %s", e.AccountRef)
		}
		sum += e.Amount
	}
	if sum != 0 {
		return newErr("UNBALANCED", "entries sum to %d, expected 0", sum)
	}
	return nil
}

// --- Service ---

type Service struct {
	db *sql.DB
}

func NewService(db *sql.DB) *Service { return &Service{db: db} }

func (s *Service) Post(ctx context.Context, req PostRequest) (*PostResult, error) {
	if err := assertBalanced(req.Entries); err != nil {
		return nil, err
	}
	fingerprint := canonicalHash(req)

	replayed, stored, err := s.claimIdempotencyKey(ctx, req.IdempotencyKey, fingerprint)
	if err != nil {
		return nil, err
	}
	if replayed {
		stored.Replayed = true
		return stored, nil
	}

	tx, err := s.db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelSerializable})
	if err != nil {
		return nil, err
	}
	defer func() { _ = tx.Rollback() }()

	result, err := s.writeTransaction(ctx, tx, req)
	if err != nil {
		_, _ = s.db.ExecContext(ctx,
			`UPDATE idempotency_records SET status = 'failed', locked_until = NULL WHERE key = $1`,
			req.IdempotencyKey)
		return nil, err
	}

	body, _ := json.Marshal(result)
	if _, err := tx.ExecContext(ctx,
		`UPDATE idempotency_records
		    SET status = 'completed', response_body = $2, transaction_ref = $3, locked_until = NULL
		  WHERE key = $1`,
		req.IdempotencyKey, body, result.TransactionRef); err != nil {
		return nil, err
	}
	if err := tx.Commit(); err != nil {
		return nil, err
	}
	return result, nil
}

func (s *Service) claimIdempotencyKey(ctx context.Context, key, fingerprint string) (bool, *PostResult, error) {
	row := s.db.QueryRowContext(ctx,
		`INSERT INTO idempotency_records (key, request_hash, status, locked_until)
		 VALUES ($1, $2, 'in_progress', now() + interval '60 seconds')
		 ON CONFLICT (key) DO UPDATE
		    SET status = 'in_progress',
		        locked_until = now() + interval '60 seconds'
		  WHERE idempotency_records.status = 'failed'
		     OR (idempotency_records.status = 'in_progress'
		         AND idempotency_records.locked_until < now())
		 RETURNING request_hash`, key, fingerprint)

	var gotHash string
	switch err := row.Scan(&gotHash); err {
	case nil:
		if gotHash != fingerprint {
			return false, nil, newErr("IDEMPOTENCY_MISMATCH", "idempotency key reused with a different request body")
		}
		return false, nil, nil
	case sql.ErrNoRows:
		// An existing record blocked the upsert; inspect it.
	default:
		return false, nil, err
	}

	var status, storedHash string
	var body []byte
	err := s.db.QueryRowContext(ctx,
		`SELECT status, request_hash, COALESCE(response_body, '{}'::jsonb)
		   FROM idempotency_records WHERE key = $1`, key).Scan(&status, &storedHash, &body)
	if err == sql.ErrNoRows {
		return false, nil, newErr("RETRY", "idempotency race, retry the request")
	}
	if err != nil {
		return false, nil, err
	}
	if storedHash != fingerprint {
		return false, nil, newErr("IDEMPOTENCY_MISMATCH", "idempotency key reused with a different request body")
	}
	if status == "completed" {
		var stored PostResult
		if err := json.Unmarshal(body, &stored); err != nil {
			return false, nil, err
		}
		return true, &stored, nil
	}
	return false, nil, newErr("IN_PROGRESS", "an identical request is currently being processed")
}

type accountRow struct {
	ID            int64
	Ref           string
	Currency      string
	AccountType   string
	AllowNegative bool
	Balance       int64
}

func (s *Service) writeTransaction(ctx context.Context, tx *sql.Tx, req PostRequest) (*PostResult, error) {
	seen := map[string]struct{}{}
	refs := make([]string, 0, len(req.Entries))
	for _, e := range req.Entries {
		if _, ok := seen[e.AccountRef]; !ok {
			seen[e.AccountRef] = struct{}{}
			refs = append(refs, e.AccountRef)
		}
	}
	// Deterministic lock order prevents deadlocks between concurrent transfers.
	sort.Strings(refs)

	rows, err := tx.QueryContext(ctx,
		`SELECT a.id, a.account_ref, a.currency, a.account_type, a.allow_negative, b.balance
		   FROM accounts a
		   JOIN account_balances b ON b.account_id = a.id
		  WHERE a.account_ref = ANY($1)
		  ORDER BY a.account_ref
		    FOR UPDATE OF b`, pgTextArray(refs))
	if err != nil {
		return nil, err
	}
	byRef := map[string]*accountRow{}
	for rows.Next() {
		var a accountRow
		if err := rows.Scan(&a.ID, &a.Ref, &a.Currency, &a.AccountType, &a.AllowNegative, &a.Balance); err != nil {
			rows.Close()
			return nil, err
		}
		byRef[a.Ref] = &a
	}
	rows.Close()
	if err := rows.Err(); err != nil {
		return nil, err
	}

	if len(byRef) != len(refs) {
		missing := []string{}
		for _, r := range refs {
			if _, ok := byRef[r]; !ok {
				missing = append(missing, r)
			}
		}
		return nil, newErr("ACCOUNT_NOT_FOUND", "unknown accounts: %s", strings.Join(missing, ", "))
	}
	for _, a := range byRef {
		if a.Currency != req.Currency {
			return nil, newErr("CURRENCY_MISMATCH", "account %s is %s, transaction is %s", a.Ref, a.Currency, req.Currency)
		}
	}

	txRef := uuid.NewString()
	meta, _ := json.Marshal(req.Metadata)
	if meta == nil {
		meta = []byte("{}")
	}

	var txID int64
	if err := tx.QueryRowContext(ctx,
		`INSERT INTO ledger_transactions
		   (transaction_ref, idempotency_key, kind, reverses_ref, description, metadata)
		 VALUES ($1, $2, $3, $4, $5, $6)
		 RETURNING id`,
		txRef, req.IdempotencyKey, req.Kind, req.ReversesRef, req.Description, meta,
	).Scan(&txID); err != nil {
		return nil, err
	}

	var sequenceNo int64
	if err := tx.QueryRowContext(ctx, `SELECT nextval('ledger_sequence')`).Scan(&sequenceNo); err != nil {
		return nil, err
	}

	deltas := map[string]int64{}
	for _, e := range req.Entries {
		deltas[e.AccountRef] += e.Amount
		acc := byRef[e.AccountRef]
		if _, err := tx.ExecContext(ctx,
			`INSERT INTO ledger_entries (transaction_id, account_id, amount, currency, sequence_no)
			 VALUES ($1, $2, $3, $4, $5)`,
			txID, acc.ID, e.Amount, req.Currency, sequenceNo); err != nil {
			return nil, err
		}
	}

	balances := map[string]int64{}
	for _, ref := range refs {
		acc := byRef[ref]
		next := acc.Balance + deltas[ref]
		if !acc.AllowNegative {
			if acc.AccountType == "liability" && next > 0 {
				return nil, newErr("INSUFFICIENT_FUNDS", "account %s would go negative", ref)
			}
			if acc.AccountType == "asset" && next < 0 {
				return nil, newErr("INSUFFICIENT_FUNDS", "account %s would go negative", ref)
			}
		}
		if _, err := tx.ExecContext(ctx,
			`UPDATE account_balances
			    SET balance = $2, last_sequence = $3, updated_at = now()
			  WHERE account_id = $1`, acc.ID, next, sequenceNo); err != nil {
			return nil, err
		}
		balances[ref] = next
	}

	var total int64
	if err := tx.QueryRowContext(ctx,
		`SELECT COALESCE(SUM(amount), 0) FROM ledger_entries WHERE transaction_id = $1`,
		txID).Scan(&total); err != nil {
		return nil, err
	}
	if total != 0 {
		return nil, newErr("UNBALANCED", "post-write invariant check failed")
	}

	return &PostResult{
		TransactionRef: txRef,
		SequenceNo:     sequenceNo,
		Replayed:       false,
		Balances:       balances,
	}, nil
}

// Reverse appends the mirror image of a transaction. Nothing is ever updated
// or deleted; the original stays visible forever.
func (s *Service) Reverse(ctx context.Context, originalRef, reason, idempotencyKey string) (*PostResult, error) {
	var alreadyReversed int
	if err := s.db.QueryRowContext(ctx,
		`SELECT count(*) FROM ledger_transactions WHERE reverses_ref = $1`, originalRef,
	).Scan(&alreadyReversed); err != nil {
		return nil, err
	}
	if alreadyReversed > 0 {
		return nil, newErr("ALREADY_REVERSED", "transaction %s is already reversed", originalRef)
	}

	rows, err := s.db.QueryContext(ctx,
		`SELECT a.account_ref, e.amount, e.currency
		   FROM ledger_transactions t
		   JOIN ledger_entries e ON e.transaction_id = t.id
		   JOIN accounts a ON a.id = e.account_id
		  WHERE t.transaction_ref = $1`, originalRef)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	var entries []EntryInput
	currency := ""
	for rows.Next() {
		var ref, cur string
		var amount int64
		if err := rows.Scan(&ref, &amount, &cur); err != nil {
			return nil, err
		}
		currency = cur
		entries = append(entries, EntryInput{AccountRef: ref, Amount: -amount})
	}
	if len(entries) == 0 {
		return nil, newErr("NOT_FOUND", "transaction %s not found", originalRef)
	}

	return s.Post(ctx, PostRequest{
		IdempotencyKey: idempotencyKey,
		Kind:           "reversal",
		Currency:       currency,
		Description:    fmt.Sprintf("reversal of %s: %s", originalRef, reason),
		ReversesRef:    &originalRef,
		Entries:        entries,
		Metadata:       map[string]interface{}{"reason": reason, "reverses": originalRef},
	})
}

// AuthoritativeBalance derives the balance from the ledger itself, ignoring
// the materialised projection. Pass asOfSequence for a point-in-time answer.
func (s *Service) AuthoritativeBalance(ctx context.Context, accountRef string, asOfSequence *int64) (int64, error) {
	var total int64
	err := s.db.QueryRowContext(ctx,
		`SELECT COALESCE(SUM(e.amount), 0)
		   FROM ledger_entries e
		   JOIN accounts a ON a.id = e.account_id
		  WHERE a.account_ref = $1
		    AND ($2::bigint IS NULL OR e.sequence_no <= $2::bigint)`,
		accountRef, asOfSequence).Scan(&total)
	return total, err
}

func pgTextArray(values []string) string {
	quoted := make([]string, len(values))
	for i, v := range values {
		quoted[i] = `"` + strings.ReplaceAll(v, `"`, `\"`) + `"`
	}
	return "{" + strings.Join(quoted, ",") + "}"
}

package payout

import (
	"context"
	"errors"
	"fmt"
	"time"

	"example.com/ledger"
)

const (
	suspenseAccount = "suspense:payout-inflight"
	bankAccount     = "bank:settlement:baghdad"
	providerFeeAcct = "expense:provider-fees"
	maxUnknownTries = 12
)

// --- Provider contract ---

type OutcomeStatus string

const (
	OutcomeSucceeded OutcomeStatus = "succeeded"
	OutcomeFailed    OutcomeStatus = "failed"
	OutcomeUnknown   OutcomeStatus = "unknown"
)

type ProviderOutcome struct {
	Status      OutcomeStatus
	ProviderRef string
	Reason      string
}

type Provider interface {
	// Authorize must be idempotent on ourRef.
	Authorize(ctx context.Context, ourRef string, amount int64, destination string) (ProviderOutcome, error)
	Capture(ctx context.Context, ourRef string) (ProviderOutcome, error)
	Void(ctx context.Context, ourRef string) (ProviderOutcome, error)
	// Lookup is the escape hatch for unknown outcomes. A nil result means the
	// provider has never seen this reference.
	Lookup(ctx context.Context, ourRef string) (*ProviderOutcome, error)
}

// --- Saga state ---

type State string

const (
	StateReserved     State = "reserved"
	StateAuthorized   State = "authorized"
	StateCaptured     State = "captured"
	StateSettled      State = "settled"
	StateCompensating State = "compensating"
	StateCompensated  State = "compensated"
	StateUnknown      State = "unknown"
	StateStuck        State = "stuck"
)

type Record struct {
	OurRef      string
	UserAccount string
	Amount      int64
	Currency    string
	Destination string
	State       State
	Attempts    int
	ProviderRef string
	LastError   string
}

type Store interface {
	Create(ctx context.Context, r Record) (*Record, error)
	Load(ctx context.Context, ourRef string) (*Record, error)
	Transition(ctx context.Context, ourRef string, from []State, to State, patch map[string]interface{}) (bool, error)
	ClaimPending(ctx context.Context, limit int, olderThan time.Duration) ([]Record, error)
}

type Logger func(event string, fields map[string]interface{})

// --- Saga ---

type Saga struct {
	ledger   *ledger.Service
	provider Provider
	store    Store
	log      Logger
}

func New(l *ledger.Service, p Provider, s Store, log Logger) *Saga {
	return &Saga{ledger: l, provider: p, store: s, log: log}
}

// Start moves money out of the user's wallet into suspense inside a single
// ledger transaction, before the provider is touched at all. If we crash after
// this point the money is still fully accounted for.
func (s *Saga) Start(ctx context.Context, r Record) (*Record, error) {
	if existing, err := s.store.Load(ctx, r.OurRef); err == nil && existing != nil {
		return existing, nil
	}

	_, err := s.ledger.Post(ctx, ledger.PostRequest{
		IdempotencyKey: "payout-reserve:" + r.OurRef,
		Kind:           "payout",
		Currency:       r.Currency,
		Description:    "reserve payout " + r.OurRef,
		Entries: []ledger.EntryInput{
			{AccountRef: r.UserAccount, Amount: r.Amount},
			{AccountRef: suspenseAccount, Amount: -r.Amount},
		},
		Metadata: map[string]interface{}{"ourRef": r.OurRef, "stage": "reserve"},
	})
	if err != nil {
		return nil, err
	}

	r.State = StateReserved
	rec, err := s.store.Create(ctx, r)
	if err != nil {
		return nil, err
	}
	s.log("payout.reserved", map[string]interface{}{"ourRef": r.OurRef, "amount": r.Amount})
	return rec, nil
}

// Advance drives one saga forward by exactly one step. Safe to call repeatedly.
func (s *Saga) Advance(ctx context.Context, ourRef string) (State, error) {
	saga, err := s.store.Load(ctx, ourRef)
	if err != nil {
		return "", err
	}
	if saga == nil {
		return "", fmt.Errorf("unknown saga %s", ourRef)
	}

	switch saga.State {
	case StateReserved:
		return s.doAuthorize(ctx, saga)
	case StateAuthorized:
		return s.doCapture(ctx, saga)
	case StateCaptured:
		return s.doSettle(ctx, saga)
	case StateUnknown:
		return s.resolveUnknown(ctx, saga)
	case StateCompensating:
		return s.doCompensate(ctx, saga)
	default:
		return saga.State, nil
	}
}

func (s *Saga) doAuthorize(ctx context.Context, saga *Record) (State, error) {
	outcome, err := s.provider.Authorize(ctx, saga.OurRef, saga.Amount, saga.Destination)
	if err != nil {
		// An error is indistinguishable from a timeout. Never assume.
		s.log("payout.authorize.unknown", map[string]interface{}{"ourRef": saga.OurRef, "error": err.Error()})
		_, _ = s.store.Transition(ctx, saga.OurRef, []State{StateReserved}, StateUnknown,
			map[string]interface{}{"last_error": err.Error()})
		return StateUnknown, nil
	}

	switch outcome.Status {
	case OutcomeSucceeded:
		_, err = s.store.Transition(ctx, saga.OurRef, []State{StateReserved}, StateAuthorized,
			map[string]interface{}{"provider_ref": outcome.ProviderRef})
		return StateAuthorized, err
	case OutcomeFailed:
		if _, err := s.store.Transition(ctx, saga.OurRef, []State{StateReserved}, StateCompensating,
			map[string]interface{}{"last_error": outcome.Reason}); err != nil {
			return "", err
		}
		saga.State = StateCompensating
		saga.LastError = outcome.Reason
		return s.doCompensate(ctx, saga)
	default:
		_, err = s.store.Transition(ctx, saga.OurRef, []State{StateReserved}, StateUnknown,
			map[string]interface{}{"provider_ref": outcome.ProviderRef})
		return StateUnknown, err
	}
}

func (s *Saga) doCapture(ctx context.Context, saga *Record) (State, error) {
	outcome, err := s.provider.Capture(ctx, saga.OurRef)
	if err != nil {
		_, _ = s.store.Transition(ctx, saga.OurRef, []State{StateAuthorized}, StateUnknown,
			map[string]interface{}{"last_error": err.Error()})
		return StateUnknown, nil
	}

	switch outcome.Status {
	case OutcomeSucceeded:
		_, err = s.store.Transition(ctx, saga.OurRef, []State{StateAuthorized}, StateCaptured,
			map[string]interface{}{"provider_ref": outcome.ProviderRef})
		return StateCaptured, err
	case OutcomeFailed:
		if _, err := s.store.Transition(ctx, saga.OurRef, []State{StateAuthorized}, StateCompensating,
			map[string]interface{}{"last_error": outcome.Reason}); err != nil {
			return "", err
		}
		saga.State = StateCompensating
		saga.LastError = outcome.Reason
		return s.doCompensate(ctx, saga)
	default:
		_, err = s.store.Transition(ctx, saga.OurRef, []State{StateAuthorized}, StateUnknown, nil)
		return StateUnknown, err
	}
}

// doSettle moves money out of suspense and against the bank account. Only now
// is the payout truly done from the ledger's point of view.
func (s *Saga) doSettle(ctx context.Context, saga *Record) (State, error) {
	fee := providerFee(saga.Amount)
	_, err := s.ledger.Post(ctx, ledger.PostRequest{
		IdempotencyKey: "payout-settle:" + saga.OurRef,
		Kind:           "payout",
		Currency:       saga.Currency,
		Description:    "settle payout " + saga.OurRef,
		Entries: []ledger.EntryInput{
			{AccountRef: suspenseAccount, Amount: saga.Amount},
			{AccountRef: bankAccount, Amount: -(saga.Amount - fee)},
			{AccountRef: providerFeeAcct, Amount: -fee},
		},
		Metadata: map[string]interface{}{
			"ourRef": saga.OurRef, "providerRef": saga.ProviderRef, "stage": "settle",
		},
	})
	if err != nil {
		return "", err
	}
	if _, err := s.store.Transition(ctx, saga.OurRef, []State{StateCaptured}, StateSettled, nil); err != nil {
		return "", err
	}
	s.log("payout.settled", map[string]interface{}{"ourRef": saga.OurRef})
	return StateSettled, nil
}

// doCompensate is not a rollback. It is a new, forward-only ledger transaction
// that returns the money to the user's wallet.
func (s *Saga) doCompensate(ctx context.Context, saga *Record) (State, error) {
	if _, err := s.provider.Void(ctx, saga.OurRef); err != nil {
		// Voiding is best effort; the reconciler catches a stuck void.
		s.log("payout.void.failed", map[string]interface{}{"ourRef": saga.OurRef, "error": err.Error()})
	}

	_, err := s.ledger.Post(ctx, ledger.PostRequest{
		IdempotencyKey: "payout-release:" + saga.OurRef,
		Kind:           "reversal",
		Currency:       saga.Currency,
		Description:    "release payout " + saga.OurRef,
		Entries: []ledger.EntryInput{
			{AccountRef: suspenseAccount, Amount: saga.Amount},
			{AccountRef: saga.UserAccount, Amount: -saga.Amount},
		},
		Metadata: map[string]interface{}{
			"ourRef": saga.OurRef, "stage": "compensate", "reason": saga.LastError,
		},
	})
	if err != nil {
		return "", err
	}
	if _, err := s.store.Transition(ctx, saga.OurRef,
		[]State{StateCompensating, StateUnknown}, StateCompensated, nil); err != nil {
		return "", err
	}
	s.log("payout.compensated", map[string]interface{}{"ourRef": saga.OurRef})
	return StateCompensated, nil
}

// resolveUnknown asks the provider what actually happened, using our own
// reference. We never guess, and we never time out into a guess.
func (s *Saga) resolveUnknown(ctx context.Context, saga *Record) (State, error) {
	outcome, err := s.provider.Lookup(ctx, saga.OurRef)
	if err != nil {
		return StateUnknown, nil // try again on the next sweep
	}

	if outcome == nil {
		// The provider has never heard of this reference: nothing happened.
		if _, err := s.store.Transition(ctx, saga.OurRef,
			[]State{StateUnknown}, StateCompensating, nil); err != nil {
			return "", err
		}
		saga.State = StateCompensating
		return s.doCompensate(ctx, saga)
	}

	switch outcome.Status {
	case OutcomeSucceeded:
		_, err = s.store.Transition(ctx, saga.OurRef, []State{StateUnknown}, StateCaptured,
			map[string]interface{}{"provider_ref": outcome.ProviderRef})
		return StateCaptured, err
	case OutcomeFailed:
		if _, err := s.store.Transition(ctx, saga.OurRef, []State{StateUnknown}, StateCompensating,
			map[string]interface{}{"last_error": outcome.Reason}); err != nil {
			return "", err
		}
		saga.State = StateCompensating
		saga.LastError = outcome.Reason
		return s.doCompensate(ctx, saga)
	}

	// Still unknown after N attempts: escalate to a human. Silence is worse
	// than a page.
	if saga.Attempts >= maxUnknownTries {
		_, _ = s.store.Transition(ctx, saga.OurRef, []State{StateUnknown}, StateStuck, nil)
		s.log("payout.stuck", map[string]interface{}{"ourRef": saga.OurRef, "attempts": saga.Attempts})
		return StateStuck, nil
	}
	return StateUnknown, nil
}

// RunSweep pushes every non-terminal saga forward. Run it on a short interval.
func (s *Saga) RunSweep(ctx context.Context, batchSize int) error {
	pending, err := s.store.ClaimPending(ctx, batchSize, 30*time.Second)
	if err != nil {
		return err
	}
	var errs []error
	for i := range pending {
		if _, err := s.Advance(ctx, pending[i].OurRef); err != nil {
			s.log("payout.sweep.error", map[string]interface{}{
				"ourRef": pending[i].OurRef, "error": err.Error(),
			})
			errs = append(errs, err)
		}
	}
	return errors.Join(errs...)
}

// providerFee is 0.9% plus 200 minor units, rounded up, capped at the amount.
func providerFee(amount int64) int64 {
	pct := (amount*9 + 999) / 1000
	fee := pct + 200
	if fee > amount {
		return amount
	}
	return fee
}