feat(loyalty): exchange EnakCoin into EnakPoint

Adds GET /customer/wallet/exchange/preview?coins= and
POST /customer/wallet/exchange (docs/prd-point-coin.md F4, K3, PC-401).

The customer exchanges a multiple of the organization's coin_amount and
gets (coins / coin_amount) x point_amount EnakPoint, approved by their PIN
(K8). A malformed amount is refused before the PIN is checked, so it costs
no attempt. In one transaction the wallet is locked, EXCHANGE_OUT takes the
EnakCoin in K9 order and EXCHANGE_IN adds the EnakPoint; the two rows share
a group, point at each other and both freeze the rate in their metadata.

The EnakPoint are split over the EnakCoin lots they came from, each part
keeping its lot's expiry and pointing back at it, so exchanging cannot
extend a balance's life. The split takes floor(coins so far x rate) per
lot, which adds up exactly because the total is a multiple of coin_amount.
EnakPoint have no validity of their own until the expiry model is decided
(N4), so the EnakCoin lot is for now the only bound.

The Idempotency-Key header (or X-Idempotency-Key) is required. A retry
with the same key is recognised under the wallet lock and replayed with
the ids and rate the first attempt froze, even if the rate has changed
since; the same key for another amount is refused.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

internal/processor/wallet_exchange_processor.go

@@ -0,0 +1,290 @@
+package processor
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/google/uuid"
+
+	"apskel-pos-be/internal/constants"
+	"apskel-pos-be/internal/entities"
+	"apskel-pos-be/internal/models"
+	"apskel-pos-be/internal/repository"
+)
+
+// ErrWalletMoveRejected wraps every reason an exchange or a transfer is refused on
+// the customer's side: the amount, the limits, the recipient or the balance. The
+// message says which.
+var ErrWalletMoveRejected = errors.New("wallet move refused")
+
+// walletMoveKeyLimit keeps a client's Idempotency-Key short enough to fit, with the
+// prefix that scopes it to the customer, in wallet_transactions.idempotency_key.
+const walletMoveKeyLimit = 50
+
+type organizationSettingsReader interface {
+	Organization(ctx context.Context, organizationID uuid.UUID) (*models.OrganizationLoyaltySettings, error)
+}
+
+// WalletExchangeProcessor exchanges EnakCoin into EnakPoint (docs/prd-point-coin.md
+// F4, K3). It is one way only; nothing turns EnakPoint back into EnakCoin.
+type WalletExchangeProcessor struct {
+	customers repository.WalletMoveRepository
+	settings  organizationSettingsReader
+	spendable spendableReader
+	pins      pinVerifier
+	wallet    *WalletProcessor
+	tx        TxRunner
+	now       func() time.Time
+}
+
+func NewWalletExchangeProcessor(customers repository.WalletMoveRepository, settings organizationSettingsReader, spendable spendableReader, pins pinVerifier, wallet *WalletProcessor, tx TxRunner) *WalletExchangeProcessor {
+	return &WalletExchangeProcessor{customers: customers, settings: settings, spendable: spendable, pins: pins, wallet: wallet, tx: tx, now: time.Now}
+}
+
+// Preview is GET /customer/wallet/exchange/preview: the organization's rate and what
+// exchanging coins would give, so the app can show it before asking for the PIN.
+func (p *WalletExchangeProcessor) Preview(ctx context.Context, customerID uuid.UUID, coins int64) (*models.WalletExchangePreview, error) {
+	customer, err := p.customers.GetCustomer(ctx, customerID)
+	if err != nil {
+		return nil, err
+	}
+	settings, err := p.settings.Organization(ctx, customer.OrganizationID)
+	if err != nil {
+		return nil, err
+	}
+	balances, err := p.spendable.SpendableBalances(ctx, customerID, p.now())
+	if err != nil {
+		return nil, err
+	}
+	rate := settings.Exchange
+	preview := &models.WalletExchangePreview{
+		CoinAmount:  rate.CoinAmount,
+		PointAmount: rate.PointAmount,
+		CoinBalance: balances[constants.WalletCurrencyCoin],
+		Coins:       coins,
+	}
+	if reason := exchangeProblem(customer, rate, coins); reason != "" {
+		preview.Reason = reason
+		return preview, nil
+	}
+	preview.Points = exchangePoints(coins, rate)
+	if coins > preview.CoinBalance {
+		preview.Reason = "not enough EnakCoin"
+		return preview, nil
+	}
+	preview.Valid = true
+	return preview, nil
+}
+
+// Exchange takes coins EnakCoin and gives the EnakPoint they are worth at the
+// organization's rate, in one transaction. The two ledger rows share a group and
+// point at each other, and both freeze the rate. Each EnakPoint lot keeps the expiry
+// of the EnakCoin lot it came from, so exchanging cannot extend a balance's life
+// (K9). The PIN approves it (K8).
+//
+// idempotencyKey is the client's Idempotency-Key: a retry with the same key returns
+// the first exchange, at the rate it was made, without moving anything again.
+func (p *WalletExchangeProcessor) Exchange(ctx context.Context, customerID uuid.UUID, coins int64, pin, idempotencyKey string, info models.CustomerPinRequestInfo) (*models.WalletExchangeResult, error) {
+	key, err := walletMoveKey(idempotencyKey)
+	if err != nil {
+		return nil, err
+	}
+	customer, err := p.customers.GetCustomer(ctx, customerID)
+	if err != nil {
+		return nil, err
+	}
+	settings, err := p.settings.Organization(ctx, customer.OrganizationID)
+	if err != nil {
+		return nil, err
+	}
+	// Refuse a malformed request before the PIN is checked, so a typo in the amount
+	// costs the customer no PIN attempt.
+	if reason := exchangeProblem(customer, settings.Exchange, coins); reason != "" {
+		return nil, fmt.Errorf("%w: %s", ErrWalletMoveRejected, reason)
+	}
+	if err := p.pins.VerifyPin(ctx, customerID, pin, PinActionExchange, info); err != nil {
+		return nil, err
+	}
+
+	outKey := fmt.Sprintf("exchange:%s:%s:out", customerID, key)
+	inKey := fmt.Sprintf("exchange:%s:%s:in", customerID, key)
+	result := &models.WalletExchangeResult{Coins: coins}
+	err = p.tx.WithTransaction(ctx, func(ctx context.Context) error {
+		if err := p.wallet.LockWallet(ctx, customerID); err != nil {
+			return err
+		}
+		rate := settings.Exchange
+		groupID, outID, inID := uuid.New(), uuid.New(), uuid.New()
+		previous, err := p.wallet.FindTransaction(ctx, outKey)
+		if err != nil {
+			return err
+		}
+		if previous != nil && previous.GroupID != nil {
+			// A retry: repeat it with the ids and the rate the first attempt froze, so
+			// both rows replay even if the rate has changed since.
+			outID, inID, groupID = previous.ID, previous.ReferenceID, *previous.GroupID
+			rate = frozenExchangeRate(previous.Metadata, rate)
+		}
+		points := exchangePoints(coins, rate)
+		metadata := entities.Metadata{
+			"coins":        coins,
+			"points":       points,
+			"coin_amount":  rate.CoinAmount,
+			"point_amount": rate.PointAmount,
+		}
+
+		out, err := p.wallet.Debit(ctx, WalletDebitInput{WalletEntry: WalletEntry{
+			TransactionID:  outID,
+			CustomerID:     customerID,
+			Currency:       constants.WalletCurrencyCoin,
+			Type:           constants.WalletTxTypeExchangeOut,
+			Amount:         coins,
+			ReferenceType:  constants.WalletRefTypeWalletTx,
+			ReferenceID:    inID,
+			GroupID:        &groupID,
+			Description:    fmt.Sprintf("Tukar %d EnakCoin ke EnakPoint", coins),
+			Metadata:       metadata,
+			IdempotencyKey: outKey,
+		}})
+		if errors.Is(err, repository.ErrWalletInsufficientBalance) {
+			return fmt.Errorf("%w: not enough EnakCoin", ErrWalletMoveRejected)
+		}
+		if err != nil {
+			return err
+		}
+		in, err := p.wallet.Credit(ctx, WalletCreditInput{
+			WalletEntry: WalletEntry{
+				TransactionID:  inID,
+				CustomerID:     customerID,
+				Currency:       constants.WalletCurrencyPoint,
+				Type:           constants.WalletTxTypeExchangeIn,
+				Amount:         points,
+				ReferenceType:  constants.WalletRefTypeWalletTx,
+				ReferenceID:    outID,
+				GroupID:        &groupID,
+				Description:    fmt.Sprintf("Dari tukar %d EnakCoin", coins),
+				Metadata:       metadata,
+				IdempotencyKey: inKey,
+			},
+			Lots: exchangeLots(out.Allocations, rate),
+		})
+		if err != nil {
+			return err
+		}
+
+		result.GroupID = groupID
+		result.Points = points
+		result.CoinAmount = rate.CoinAmount
+		result.PointAmount = rate.PointAmount
+		result.Lots = movedLots(in.Lots)
+		result.Replayed = out.Replayed
+		return nil
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	balances, err := p.spendable.SpendableBalances(ctx, customerID, p.now())
+	if err != nil {
+		return nil, err
+	}
+	result.CoinBalance = balances[constants.WalletCurrencyCoin]
+	result.PointBalance = balances[constants.WalletCurrencyPoint]
+	return result, nil
+}
+
+// exchangeProblem says why coins cannot be exchanged, or "" when they can as far as
+// the request goes. The balance is checked under the wallet lock.
+func exchangeProblem(customer *repository.WalletMoveCustomer, rate models.LoyaltyExchangeSettings, coins int64) string {
+	switch {
+	case !customer.IsActive:
+		return "the customer is not active"
+	case coins <= 0:
+		return "the number of EnakCoin must be positive"
+	case rate.CoinAmount <= 0 || rate.PointAmount <= 0:
+		return "exchange is not available"
+	case coins%rate.CoinAmount != 0:
+		// Otherwise part of the EnakCoin would be lost to rounding (F4).
+		return fmt.Sprintf("EnakCoin are exchanged in multiples of %d", rate.CoinAmount)
+	}
+	return ""
+}
+
+// exchangePoints is (coins / coin_amount) × point_amount, for coins that are a
+// multiple of coin_amount.
+func exchangePoints(coins int64, rate models.LoyaltyExchangeSettings) int64 {
+	return coins / rate.CoinAmount * rate.PointAmount
+}
+
+// exchangeLots splits the EnakPoint of an exchange over the EnakCoin lots it took,
+// so each part keeps the expiry of its lot and points back at it (K9). The share of
+// a lot is the difference of floor(coins so far × point_amount / coin_amount) before
+// and after it, which adds up exactly because the total is a multiple of
+// coin_amount. A lot too small to earn a whole EnakPoint on its own gives none.
+//
+// F4 caps the expiry at now + the EnakPoint validity as well. EnakPoint have no
+// validity until the expiry model is decided (F12, note N4), so for now the EnakCoin
+// lot's expiry is the only bound.
+func exchangeLots(allocations []WalletAllocation, rate models.LoyaltyExchangeSettings) []WalletLotInput {
+	var lots []WalletLotInput
+	var coinsSoFar int64
+	for _, a := range allocations {
+		before := coinsSoFar * rate.PointAmount / rate.CoinAmount
+		coinsSoFar += a.Amount
+		points := coinsSoFar*rate.PointAmount/rate.CoinAmount - before
+		if points == 0 {
+			continue
+		}
+		lotID := a.LotID
+		lots = append(lots, WalletLotInput{Amount: points, ExpiresAt: a.ExpiresAt, OriginLotID: &lotID})
+	}
+	return lots
+}
+
+// frozenExchangeRate reads the rate an exchange was made at from its ledger row.
+func frozenExchangeRate(metadata entities.Metadata, fallback models.LoyaltyExchangeSettings) models.LoyaltyExchangeSettings {
+	coinAmount, ok1 := metadataInt(metadata, "coin_amount")
+	pointAmount, ok2 := metadataInt(metadata, "point_amount")
+	if !ok1 || !ok2 || coinAmount <= 0 || pointAmount <= 0 {
+		return fallback
+	}
+	return models.LoyaltyExchangeSettings{CoinAmount: coinAmount, PointAmount: pointAmount}
+}
+
+// metadataInt reads a whole number from metadata that may have been through JSONB,
+// which gives numbers back as float64.
+func metadataInt(metadata entities.Metadata, key string) (int64, bool) {
+	switch v := metadata[key].(type) {
+	case float64:
+		return int64(v), true
+	case int64:
+		return v, true
+	case int:
+		return int64(v), true
+	}
+	return 0, false
+}
+
+func movedLots(lots []entities.WalletLot) []models.WalletMovedLot {
+	out := make([]models.WalletMovedLot, 0, len(lots))
+	for _, lot := range lots {
+		out = append(out, models.WalletMovedLot{Amount: lot.OriginalAmount, ExpiresAt: lot.ExpiresAt})
+	}
+	return out
+}
+
+// walletMoveKey checks the client's Idempotency-Key, which exchanges and transfers
+// require (F4, F5).
+func walletMoveKey(key string) (string, error) {
+	key = strings.TrimSpace(key)
+	if key == "" {
+		return "", fmt.Errorf("%w: the Idempotency-Key header is required", ErrWalletMoveRejected)
+	}
+	if len(key) > walletMoveKeyLimit {
+		return "", fmt.Errorf("%w: the Idempotency-Key header must be at most %d characters", ErrWalletMoveRejected, walletMoveKeyLimit)
+	}
+	return key, nil
+}

internal/processor/wallet_exchange_processor_test.go

@@ -0,0 +1,305 @@
+package processor
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"apskel-pos-be/internal/constants"
+	"apskel-pos-be/internal/entities"
+	"apskel-pos-be/internal/models"
+	"apskel-pos-be/internal/repository"
+)
+
+// walletMoveEnv runs exchanges and transfers on the in-memory wallet, so every test
+// also checks the §7.5 invariants when it ends.
+type walletMoveEnv struct {
+	*walletTestEnv
+	customers *walletMoveRepoFake
+	settings  *models.OrganizationLoyaltySettings
+	pins      *movePinFake
+}
+
+func newWalletMoveEnv(t *testing.T) *walletMoveEnv {
+	e := &walletMoveEnv{
+		walletTestEnv: newWalletTestEnv(t),
+		customers:     &walletMoveRepoFake{byID: map[uuid.UUID]*repository.WalletMoveCustomer{}},
+		settings: &models.OrganizationLoyaltySettings{
+			PointValue: 1,
+			Exchange:   models.LoyaltyExchangeSettings{CoinAmount: 1, PointAmount: 1},
+			Transfer:   models.LoyaltyTransferSettings{Enabled: true, MinAmount: 1},
+		},
+		pins: &movePinFake{good: "482913"},
+	}
+	return e
+}
+
+// member adds an active customer of the organization.
+func (e *walletMoveEnv) member(name, phone string) uuid.UUID {
+	id := e.customer()
+	e.customers.byID[id] = &repository.WalletMoveCustomer{
+		ID: id, OrganizationID: e.org, Name: name, PhoneNumber: &phone, IsActive: true,
+	}
+	return id
+}
+
+func (e *walletMoveEnv) exchanges() *WalletExchangeProcessor {
+	p := NewWalletExchangeProcessor(e.customers, e, e, e.pins, e.p, txRunnerFake{})
+	p.now = func() time.Time { return e.now }
+	return p
+}
+
+func (e *walletMoveEnv) Organization(context.Context, uuid.UUID) (*models.OrganizationLoyaltySettings, error) {
+	s := *e.settings
+	return &s, nil
+}
+
+// SpendableBalances sums the unexpired lots, as the real query does.
+func (e *walletMoveEnv) SpendableBalances(_ context.Context, customerID uuid.UUID, asOf time.Time) (map[string]int64, error) {
+	out := map[string]int64{}
+	for _, lot := range e.repo.lots {
+		if lot.CustomerID == customerID && (lot.ExpiresAt == nil || lot.ExpiresAt.After(asOf)) {
+			out[lot.Currency] += lot.RemainingAmount
+		}
+	}
+	return out, nil
+}
+
+// earnCoins gives a customer an EnakCoin lot.
+func (e *walletMoveEnv) earnCoins(t *testing.T, customerID uuid.UUID, amount int64, expiresAt *time.Time) *WalletResult {
+	t.Helper()
+	in := earn(customerID, amount, expiresAt)
+	in.Currency = constants.WalletCurrencyCoin
+	return e.credit(t, in)
+}
+
+func (e *walletMoveEnv) coinBalance(t *testing.T, customerID uuid.UUID) int64 {
+	t.Helper()
+	w, err := e.repo.GetWallet(e.ctx, customerID)
+	require.NoError(t, err)
+	return w.CoinBalance
+}
+
+type walletMoveRepoFake struct {
+	byID map[uuid.UUID]*repository.WalletMoveCustomer
+}
+
+func (f *walletMoveRepoFake) GetCustomer(_ context.Context, id uuid.UUID) (*repository.WalletMoveCustomer, error) {
+	c, ok := f.byID[id]
+	if !ok {
+		return nil, repository.ErrWalletNotFound
+	}
+	copied := *c
+	return &copied, nil
+}
+
+// movePinFake accepts one PIN and records the actions it was asked to approve.
+type movePinFake struct {
+	good    string
+	err     error
+	actions []PinAction
+}
+
+func (f *movePinFake) VerifyPin(_ context.Context, _ uuid.UUID, pin string, action PinAction, _ models.CustomerPinRequestInfo) error {
+	f.actions = append(f.actions, action)
+	if f.err != nil {
+		return f.err
+	}
+	if pin != f.good {
+		return &PinError{Code: PinErrInvalid, RemainingAttempts: 4}
+	}
+	return nil
+}
+
+func TestWalletExchange_DefaultRateIsOneToOne(t *testing.T) {
+	e := newWalletMoveEnv(t)
+	c := e.member("Budi Santoso", "081234561234")
+	e.earnCoins(t, c, 50, nil)
+
+	res, err := e.exchanges().Exchange(e.ctx, c, 50, "482913", "key-1", models.CustomerPinRequestInfo{})
+	require.NoError(t, err)
+
+	assert.Equal(t, int64(50), res.Points)
+	assert.Equal(t, int64(0), res.CoinBalance)
+	assert.Equal(t, int64(50), res.PointBalance)
+	assert.Equal(t, []PinAction{PinActionExchange}, e.pins.actions)
+
+	out, in := e.repo.transactions[1], e.repo.transactions[2]
+	assert.Equal(t, constants.WalletTxTypeExchangeOut, out.Type)
+	assert.Equal(t, constants.WalletCurrencyCoin, out.Currency)
+	assert.Equal(t, int64(-50), out.Amount)
+	assert.Equal(t, constants.WalletTxTypeExchangeIn, in.Type)
+	assert.Equal(t, constants.WalletCurrencyPoint, in.Currency)
+	assert.Equal(t, int64(50), in.Amount)
+	// The pair shares a group and each row points at the other (§8.1).
+	assert.Equal(t, *out.GroupID, *in.GroupID)
+	assert.Equal(t, res.GroupID, *out.GroupID)
+	assert.Equal(t, in.ID, out.ReferenceID)
+	assert.Equal(t, out.ID, in.ReferenceID)
+	assert.Equal(t, "Tukar 50 EnakCoin ke EnakPoint", out.Description)
+	assert.Equal(t, "Dari tukar 50 EnakCoin", in.Description)
+	// Both rows freeze the rate.
+	for _, row := range []*entities.WalletTransaction{out, in} {
+		assert.Equal(t, int64(1), row.Metadata["coin_amount"])
+		assert.Equal(t, int64(1), row.Metadata["point_amount"])
+	}
+}
+
+func TestWalletExchange_TenCoinsForThreePoints(t *testing.T) {
+	e := newWalletMoveEnv(t)
+	e.settings.Exchange = models.LoyaltyExchangeSettings{CoinAmount: 10, PointAmount: 3}
+	c := e.member("Budi", "081234561234")
+	e.earnCoins(t, c, 35, nil)
+
+	preview, err := e.exchanges().Preview(e.ctx, c, 30)
+	require.NoError(t, err)
+	assert.True(t, preview.Valid)
+	assert.Equal(t, int64(9), preview.Points)
+	assert.Equal(t, int64(35), preview.CoinBalance)
+
+	res, err := e.exchanges().Exchange(e.ctx, c, 30, "482913", "key-1", models.CustomerPinRequestInfo{})
+	require.NoError(t, err)
+	assert.Equal(t, int64(9), res.Points)
+	assert.Equal(t, int64(5), res.CoinBalance)
+	assert.Equal(t, int64(9), res.PointBalance)
+}
+
+func TestWalletExchange_RefusesAmountsThatAreNotAMultiple(t *testing.T) {
+	e := newWalletMoveEnv(t)
+	e.settings.Exchange = models.LoyaltyExchangeSettings{CoinAmount: 10, PointAmount: 3}
+	c := e.member("Budi", "081234561234")
+	e.earnCoins(t, c, 50, nil)
+
+	preview, err := e.exchanges().Preview(e.ctx, c, 25)
+	require.NoError(t, err)
+	assert.False(t, preview.Valid)
+	assert.Contains(t, preview.Reason, "multiples of 10")
+
+	_, err = e.exchanges().Exchange(e.ctx, c, 25, "482913", "key-1", models.CustomerPinRequestInfo{})
+	assert.ErrorIs(t, err, ErrWalletMoveRejected)
+	// Refused before the PIN, so a typo costs no attempt.
+	assert.Empty(t, e.pins.actions)
+	assert.Equal(t, int64(50), e.coinBalance(t, c))
+
+	_, err = e.exchanges().Exchange(e.ctx, c, 0, "482913", "key-2", models.CustomerPinRequestInfo{})
+	assert.ErrorIs(t, err, ErrWalletMoveRejected)
+}
+
+func TestWalletExchange_NeverOutlivesTheCoinLot(t *testing.T) {
+	e := newWalletMoveEnv(t)
+	e.settings.Exchange = models.LoyaltyExchangeSettings{CoinAmount: 10, PointAmount: 3}
+	c := e.member("Budi", "081234561234")
+	soon, later := e.at(24*time.Hour), e.at(48*time.Hour)
+	first := e.earnCoins(t, c, 15, soon)
+	second := e.earnCoins(t, c, 15, later)
+	third := e.earnCoins(t, c, 10, nil)
+
+	// 40 EnakCoin take 15 from the lot expiring soonest, 15 from the next and 10 from
+	// the one that never expires, giving 12 EnakPoint split 4 + 5 + 3.
+	res, err := e.exchanges().Exchange(e.ctx, c, 40, "482913", "key-1", models.CustomerPinRequestInfo{})
+	require.NoError(t, err)
+	require.Len(t, res.Lots, 3)
+	assert.Equal(t, models.WalletMovedLot{Amount: 4, ExpiresAt: soon}, res.Lots[0])
+	assert.Equal(t, models.WalletMovedLot{Amount: 5, ExpiresAt: later}, res.Lots[1])
+	assert.Equal(t, models.WalletMovedLot{Amount: 3}, res.Lots[2])
+
+	origins := map[uuid.UUID]*time.Time{}
+	for _, lot := range e.repo.lots {
+		if lot.Currency == constants.WalletCurrencyPoint {
+			require.NotNil(t, lot.OriginLotID)
+			origins[*lot.OriginLotID] = lot.ExpiresAt
+		}
+	}
+	assert.Equal(t, map[uuid.UUID]*time.Time{
+		first.Lots[0].ID: soon, second.Lots[0].ID: later, third.Lots[0].ID: nil,
+	}, origins)
+	for _, lot := range e.repo.lots {
+		if lot.Currency != constants.WalletCurrencyPoint || lot.OriginLotID == nil {
+			continue
+		}
+		for _, coinLot := range e.repo.lots {
+			if coinLot.ID == *lot.OriginLotID && coinLot.ExpiresAt != nil {
+				require.NotNil(t, lot.ExpiresAt, "a lot that expires cannot become one that does not")
+				assert.False(t, lot.ExpiresAt.After(*coinLot.ExpiresAt))
+			}
+		}
+	}
+}
+
+func TestWalletExchange_LotTooSmallForAWholePointGivesNone(t *testing.T) {
+	e := newWalletMoveEnv(t)
+	e.settings.Exchange = models.LoyaltyExchangeSettings{CoinAmount: 10, PointAmount: 1}
+	c := e.member("Budi", "081234561234")
+	e.earnCoins(t, c, 5, e.at(time.Hour))
+	e.earnCoins(t, c, 5, nil)
+
+	// Neither lot is worth a whole EnakPoint alone; the one that completes the 10
+	// carries it.
+	res, err := e.exchanges().Exchange(e.ctx, c, 10, "482913", "key-1", models.CustomerPinRequestInfo{})
+	require.NoError(t, err)
+	assert.Equal(t, []models.WalletMovedLot{{Amount: 1}}, res.Lots)
+}
+
+func TestWalletExchange_NotEnoughCoins(t *testing.T) {
+	e := newWalletMoveEnv(t)
+	c := e.member("Budi", "081234561234")
+	e.earnCoins(t, c, 5, nil)
+
+	preview, err := e.exchanges().Preview(e.ctx, c, 6)
+	require.NoError(t, err)
+	assert.False(t, preview.Valid)
+
+	_, err = e.exchanges().Exchange(e.ctx, c, 6, "482913", "key-1", models.CustomerPinRequestInfo{})
+	assert.ErrorIs(t, err, ErrWalletMoveRejected)
+	assert.Equal(t, int64(5), e.coinBalance(t, c))
+}
+
+func TestWalletExchange_WrongPinMovesNothing(t *testing.T) {
+	e := newWalletMoveEnv(t)
+	c := e.member("Budi", "081234561234")
+	e.earnCoins(t, c, 5, nil)
+
+	_, err := e.exchanges().Exchange(e.ctx, c, 5, "000000", "key-1", models.CustomerPinRequestInfo{})
+	var pinErr *PinError
+	require.True(t, errors.As(err, &pinErr))
+	assert.Equal(t, PinErrInvalid, pinErr.Code)
+	assert.Equal(t, int64(5), e.coinBalance(t, c))
+}
+
+func TestWalletExchange_RetryReturnsTheFirstExchangeAtItsRate(t *testing.T) {
+	e := newWalletMoveEnv(t)
+	c := e.member("Budi", "081234561234")
+	e.earnCoins(t, c, 100, nil)
+
+	first, err := e.exchanges().Exchange(e.ctx, c, 40, "482913", "key-1", models.CustomerPinRequestInfo{})
+	require.NoError(t, err)
+	// The rate changes before the retry arrives; the retry still gets the first result.
+	e.settings.Exchange = models.LoyaltyExchangeSettings{CoinAmount: 1, PointAmount: 2}
+	again, err := e.exchanges().Exchange(e.ctx, c, 40, "482913", "key-1", models.CustomerPinRequestInfo{})
+	require.NoError(t, err)
+
+	assert.True(t, again.Replayed)
+	assert.Equal(t, first.GroupID, again.GroupID)
+	assert.Equal(t, int64(40), again.Points)
+	assert.Equal(t, int64(60), e.coinBalance(t, c))
+	assert.Len(t, e.repo.transactions, 3)
+
+	// The same key for a different amount is not a retry.
+	_, err = e.exchanges().Exchange(e.ctx, c, 20, "482913", "key-1", models.CustomerPinRequestInfo{})
+	assert.ErrorIs(t, err, ErrWalletIdempotencyConflict)
+}
+
+func TestWalletExchange_RequiresAnIdempotencyKey(t *testing.T) {
+	e := newWalletMoveEnv(t)
+	c := e.member("Budi", "081234561234")
+	e.earnCoins(t, c, 5, nil)
+
+	_, err := e.exchanges().Exchange(e.ctx, c, 5, "482913", " ", models.CustomerPinRequestInfo{})
+	assert.ErrorIs(t, err, ErrWalletMoveRejected)
+	assert.Empty(t, e.pins.actions)
+}

internal/processor/wallet_move_db_test.go

@@ -0,0 +1,94 @@
+package processor
+
+import (
+	"context"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"gorm.io/driver/postgres"
+	"gorm.io/gorm"
+	"gorm.io/gorm/logger"
+
+	"apskel-pos-be/internal/constants"
+	"apskel-pos-be/internal/models"
+	"apskel-pos-be/internal/repository"
+)
+
+// fixedOrganizationSettings serves the same organization settings to every caller.
+type fixedOrganizationSettings struct{ s models.OrganizationLoyaltySettings }
+
+func (f fixedOrganizationSettings) Organization(context.Context, uuid.UUID) (*models.OrganizationLoyaltySettings, error) {
+	s := f.s
+	return &s, nil
+}
+
+// walletMoveDB opens TEST_DATABASE_URL and creates an organization with two customers,
+// removed again when the test ends. See internal/repository/wallet_repository_test.go.
+func walletMoveDB(t *testing.T) (db *gorm.DB, org, a, b uuid.UUID) {
+	t.Helper()
+	dsn := os.Getenv("TEST_DATABASE_URL")
+	if dsn == "" {
+		t.Skip("TEST_DATABASE_URL not set")
+	}
+	db, err := gorm.Open(postgres.Open(dsn), &gorm.Config{Logger: logger.Default.LogMode(logger.Silent)})
+	require.NoError(t, err)
+
+	org, a, b = uuid.New(), uuid.New(), uuid.New()
+	phoneA, phoneB := "08"+a.String()[:10], "08"+b.String()[:10]
+	require.NoError(t, db.Exec(`INSERT INTO organizations (id, name, plan_type) VALUES (?, 'wallet move test', 'basic')`, org).Error)
+	require.NoError(t, db.Exec(`INSERT INTO customers (id, organization_id, name, phone_number) VALUES (?, ?, 'Anita', ?), (?, ?, 'Budi Santoso', ?)`,
+		a, org, phoneA, b, org, phoneB).Error)
+	customers := []uuid.UUID{a, b}
+	t.Cleanup(func() {
+		db.Exec(`DELETE FROM wallet_lot_allocations WHERE lot_id IN (SELECT id FROM wallet_lots WHERE customer_id IN ?)`, customers)
+		db.Exec(`DELETE FROM wallet_lots WHERE customer_id IN ? AND origin_lot_id IS NOT NULL`, customers)
+		db.Exec(`DELETE FROM wallet_lots WHERE customer_id IN ?`, customers)
+		db.Exec(`DELETE FROM wallet_transactions WHERE customer_id IN ?`, customers)
+		db.Exec(`DELETE FROM customer_wallets WHERE customer_id IN ?`, customers)
+		db.Exec(`DELETE FROM customers WHERE id IN ?`, customers)
+		db.Exec(`DELETE FROM organizations WHERE id = ?`, org)
+	})
+	return db, org, a, b
+}
+
+func TestWalletExchange_AgainstPostgres(t *testing.T) {
+	db, _, a, _ := walletMoveDB(t)
+	wallet := NewWalletProcessor(repository.NewWalletRepository(db))
+	txm := repository.NewTxManager(db)
+	settings := fixedOrganizationSettings{models.OrganizationLoyaltySettings{
+		Exchange: models.LoyaltyExchangeSettings{CoinAmount: 10, PointAmount: 3},
+	}}
+	p := NewWalletExchangeProcessor(repository.NewWalletMoveRepository(db), settings, repository.NewWalletQueryRepository(db),
+		&movePinFake{good: "482913"}, wallet, txm)
+
+	expiry := time.Now().Add(24 * time.Hour).Truncate(time.Second)
+	require.NoError(t, txm.WithTransaction(context.Background(), func(ctx context.Context) error {
+		in := earn(a, 30, &expiry)
+		in.Currency = constants.WalletCurrencyCoin
+		_, err := wallet.Credit(ctx, in)
+		return err
+	}))
+
+	res, err := p.Exchange(context.Background(), a, 20, "482913", "db-key", models.CustomerPinRequestInfo{})
+	require.NoError(t, err)
+	assert.Equal(t, int64(6), res.Points)
+	assert.Equal(t, int64(10), res.CoinBalance)
+	assert.Equal(t, int64(6), res.PointBalance)
+	require.Len(t, res.Lots, 1)
+	require.NotNil(t, res.Lots[0].ExpiresAt)
+	assert.True(t, res.Lots[0].ExpiresAt.Equal(expiry))
+
+	// The retry reads the frozen rate back out of JSONB and replays.
+	again, err := p.Exchange(context.Background(), a, 20, "482913", "db-key", models.CustomerPinRequestInfo{})
+	require.NoError(t, err)
+	assert.True(t, again.Replayed)
+	assert.Equal(t, int64(6), again.Points)
+
+	var rows int64
+	require.NoError(t, db.Raw(`SELECT COUNT(*) FROM wallet_transactions WHERE group_id = ?`, res.GroupID).Scan(&rows).Error)
+	assert.Equal(t, int64(2), rows)
+}

internal/processor/wallet_processor.go

@@ -142,6 +142,13 @@ func (p *WalletProcessor) LockWallets(ctx context.Context, a, b uuid.UUID) error
 	return err
 }
 
+// FindTransaction returns the row written under an idempotency key, or nil when the
+// key is new. Read it under the wallet lock: a flow that writes several rows uses it
+// to recognise a retry before checking limits the first attempt already counts in.
+func (p *WalletProcessor) FindTransaction(ctx context.Context, idempotencyKey string) (*entities.WalletTransaction, error) {
+	return p.repo.GetTransactionByIdempotencyKey(ctx, idempotencyKey)
+}
+
 // Credit adds Amount to the wallet and creates its lots.
 func (p *WalletProcessor) Credit(ctx context.Context, in WalletCreditInput) (*WalletResult, error) {
 	if err := validateWalletEntry(&in.WalletEntry, true); err != nil {