apskel-pos-backend/internal/processor/wallet_processor.go

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/repository"
)

var (
	// ErrWalletInvalidEntry wraps every rejection of an entry that breaks the rules in
	// docs/prd-point-coin.md §8.1. The database enforces most of them too; checking
	// here first gives callers a readable error instead of a constraint name.
	ErrWalletInvalidEntry = errors.New("wallet: invalid entry")
	// ErrWalletIdempotencyConflict means an idempotency key was reused for a different
	// operation. Retrying the same operation with the same key is not a conflict.
	ErrWalletIdempotencyConflict = errors.New("wallet: idempotency key already used for a different operation")
)

// WalletEntry is what every ledger row needs, whichever way it moves the balance.
// Which of the optional fields a type requires is listed in §8.1.
type WalletEntry struct {
	// Optional. Set it when another row must reference this one before it exists, as
	// the two rows of an exchange or a transfer do.
	TransactionID uuid.UUID

	CustomerID uuid.UUID
	Currency   string
	Type       string
	// Always positive: Credit adds it, Debit takes it away.
	Amount int64

	ReferenceType string
	ReferenceID   uuid.UUID

	GroupID                *uuid.UUID
	CounterpartyCustomerID *uuid.UUID
	ReversesTransactionID  *uuid.UUID
	OutletID               *uuid.UUID
	CreatedByUser          *uuid.UUID
	Reason                 *string

	Description string
	Metadata    entities.Metadata
	// Optional. A retry with the same key returns the first result without moving
	// anything again.
	IdempotencyKey string
}

// WalletLotInput is one lot a credit creates.
type WalletLotInput struct {
	Amount int64
	// Nil means the lot never expires.
	ExpiresAt *time.Time
	// The lot this one was carried over from, for transfers, exchanges and refunds.
	OriginLotID *uuid.UUID
}

type WalletCreditInput struct {
	WalletEntry
	// How the credit is split into lots. Their amounts must add up to Amount. Leave
	// empty for a single lot that never expires.
	Lots []WalletLotInput
}

type WalletDebitInput struct {
	WalletEntry
	// Lots to draw from first, in this order, before falling back to the K9 order.
	// A reversal names the lots its EARN created (F10), and the expiry job names the
	// lot that expired. These lots are used even if they have already expired.
	PreferredLotIDs []uuid.UUID
}

// WalletAllocation is how much a debit took from one lot. It carries the lot's
// expiry, so a transfer or exchange can give the receiving lot the same expiry (K9).
type WalletAllocation struct {
	LotID     uuid.UUID
	Amount    int64
	ExpiresAt *time.Time
}

type WalletResult struct {
	// Nil only when DebitUpTo found nothing to take.
	Transaction *entities.WalletTransaction
	// The lots a credit created.
	Lots []entities.WalletLot
	// The lots a debit drew from, in the order they were used.
	Allocations []WalletAllocation
	// What DebitUpTo could not take because the balance ran out.
	Shortfall int64
	// True when the idempotency key had already been used and nothing moved.
	Replayed bool
}

// CarryOver turns a debit's allocations into lots for the receiving side of a
// transfer or exchange. Each lot keeps the expiry of the lot it came from and points
// back at it, so a balance cannot be kept alive by moving it around (K9).
func (r *WalletResult) CarryOver() []WalletLotInput {
	lots := make([]WalletLotInput, 0, len(r.Allocations))
	for _, a := range r.Allocations {
		lotID := a.LotID
		lots = append(lots, WalletLotInput{Amount: a.Amount, ExpiresAt: a.ExpiresAt, OriginLotID: &lotID})
	}
	return lots
}

// WalletProcessor is the only code allowed to change a wallet balance. Every change
// writes the balance, the ledger row and the lots or allocations together, which is
// what keeps SUM(ledger) = balance = SUM(lot remaining) (§7.5).
//
// Every method must run inside a transaction from TxManager, and the repository
// refuses otherwise. Each method locks the customer's wallet itself, so a single-wallet
// caller needs nothing more. A caller touching two wallets, such as a transfer, must
// call LockWallets first so the locks are always taken in the same order.
type WalletProcessor struct {
	repo repository.WalletRepository
	now  func() time.Time
}

func NewWalletProcessor(repo repository.WalletRepository) *WalletProcessor {
	return &WalletProcessor{repo: repo, now: time.Now}
}

// LockWallet locks one customer's wallet, creating it if needed. Credit and Debit do
// this themselves; call it when something must be read under the lock first.
func (p *WalletProcessor) LockWallet(ctx context.Context, customerID uuid.UUID) error {
	_, err := p.repo.LockWallet(ctx, customerID)
	return err
}

// LockWallets locks two customers' wallets in a fixed order. Call it before touching
// both wallets in one transaction.
func (p *WalletProcessor) LockWallets(ctx context.Context, a, b uuid.UUID) error {
	_, _, err := p.repo.LockWallets(ctx, a, b)
	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)
}

// ExpireLot takes what is left in a lot whose expiry has passed at asOf, as an EXPIRE
// row pointing at the lot (F12). It locks the wallet before reading the lot again, so
// it never races a payment for the same balance, and the key expire:{lot_id} makes a
// second run, on this instance or another, take nothing more. describe gives the
// row's description for the amount taken. It returns nil when
// there is nothing to take: the lot is empty, not due, or already expired.
func (p *WalletProcessor) ExpireLot(ctx context.Context, lotID uuid.UUID, describe func(amount int64) string, asOf time.Time) (*WalletResult, error) {
	lot, err := p.getLot(ctx, lotID)
	if err != nil {
		return nil, err
	}
	if _, err := p.repo.LockWallet(ctx, lot.CustomerID); err != nil {
		return nil, err
	}
	// Read again under the lock: a payment may have used it up meanwhile.
	if lot, err = p.getLot(ctx, lotID); err != nil {
		return nil, err
	}
	if lot.RemainingAmount == 0 || lot.ExpiresAt == nil || lot.ExpiresAt.After(asOf) {
		return nil, nil
	}
	return p.Debit(ctx, WalletDebitInput{
		WalletEntry: WalletEntry{
			CustomerID:     lot.CustomerID,
			Currency:       lot.Currency,
			Type:           constants.WalletTxTypeExpire,
			Amount:         lot.RemainingAmount,
			ReferenceType:  constants.WalletRefTypeLot,
			ReferenceID:    lot.ID,
			Description:    describe(lot.RemainingAmount),
			Metadata:       entities.Metadata{"expires_at": lot.ExpiresAt.UTC().Format(time.RFC3339)},
			IdempotencyKey: "expire:" + lot.ID.String(),
		},
		// Exactly what the lot holds, from the lot itself, even though it has expired.
		PreferredLotIDs: []uuid.UUID{lot.ID},
	})
}

func (p *WalletProcessor) getLot(ctx context.Context, lotID uuid.UUID) (*entities.WalletLot, error) {
	lots, err := p.repo.GetLotsByIDs(ctx, []uuid.UUID{lotID})
	if err != nil {
		return nil, err
	}
	if len(lots) == 0 {
		return nil, fmt.Errorf("%w: lot %s does not exist", ErrWalletInvalidEntry, lotID)
	}
	return &lots[0], nil
}

// 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 {
		return nil, err
	}
	lots := in.Lots
	if len(lots) == 0 {
		lots = []WalletLotInput{{Amount: in.Amount}}
	}
	var total int64
	for _, lot := range lots {
		if lot.Amount <= 0 {
			return nil, fmt.Errorf("%w: lot amount must be positive, got %d", ErrWalletInvalidEntry, lot.Amount)
		}
		total += lot.Amount
	}
	if total != in.Amount {
		return nil, fmt.Errorf("%w: lots add up to %d, not %d", ErrWalletInvalidEntry, total, in.Amount)
	}

	wallet, err := p.repo.LockWallet(ctx, in.CustomerID)
	if err != nil {
		return nil, err
	}
	if replay, err := p.replay(ctx, &in.WalletEntry, true, true); replay != nil || err != nil {
		return replay, err
	}

	balance, err := p.repo.AddBalance(ctx, in.CustomerID, in.Currency, in.Amount)
	if err != nil {
		return nil, err
	}
	walletTx := newWalletTransaction(wallet, &in.WalletEntry, in.Amount, balance, nil)
	if err := p.repo.CreateTransaction(ctx, walletTx); err != nil {
		return nil, fmt.Errorf("failed to create wallet transaction: %w", err)
	}

	result := &WalletResult{Transaction: walletTx}
	for _, lotIn := range lots {
		lot := entities.WalletLot{
			OrganizationID:      wallet.OrganizationID,
			CustomerID:          in.CustomerID,
			Currency:            in.Currency,
			SourceTransactionID: walletTx.ID,
			OriginLotID:         lotIn.OriginLotID,
			OriginalAmount:      lotIn.Amount,
			RemainingAmount:     lotIn.Amount,
			ExpiresAt:           lotIn.ExpiresAt,
		}
		if err := p.repo.CreateLot(ctx, &lot); err != nil {
			return nil, fmt.Errorf("failed to create wallet lot: %w", err)
		}
		result.Lots = append(result.Lots, lot)
	}
	return result, nil
}

// Debit takes exactly Amount from the wallet, or nothing at all with
// repository.ErrWalletInsufficientBalance if the usable balance is short.
func (p *WalletProcessor) Debit(ctx context.Context, in WalletDebitInput) (*WalletResult, error) {
	return p.debit(ctx, in, false)
}

// DebitUpTo takes as much of Amount as the wallet has and reports the rest as
// Shortfall. It is for reversing earnings the customer has already spent (F10, Q3).
// When there is nothing to take, no ledger row is written and Transaction is nil;
// such a call leaves no trace, so a retry with the same key takes whatever the
// balance holds by then.
func (p *WalletProcessor) DebitUpTo(ctx context.Context, in WalletDebitInput) (*WalletResult, error) {
	return p.debit(ctx, in, true)
}

func (p *WalletProcessor) debit(ctx context.Context, in WalletDebitInput, upTo bool) (*WalletResult, error) {
	if err := validateWalletEntry(&in.WalletEntry, false); err != nil {
		return nil, err
	}
	wallet, err := p.repo.LockWallet(ctx, in.CustomerID)
	if err != nil {
		return nil, err
	}
	// DebitUpTo may have taken less than asked, so the amount cannot be compared.
	if replay, err := p.replay(ctx, &in.WalletEntry, false, !upTo); replay != nil || err != nil {
		return replay, err
	}

	lots, err := p.spendableLots(ctx, &in)
	if err != nil {
		return nil, err
	}

	var available int64
	for _, lot := range lots {
		available += lot.RemainingAmount
	}
	take := in.Amount
	if available < take {
		if !upTo {
			return nil, repository.ErrWalletInsufficientBalance
		}
		take = available
	}
	result := &WalletResult{Shortfall: in.Amount - take}
	if take == 0 {
		return result, nil
	}

	var metadata entities.Metadata
	if upTo {
		metadata = entities.Metadata{"requested_amount": in.Amount, "shortfall": result.Shortfall}
	}

	balance, err := p.repo.AddBalance(ctx, in.CustomerID, in.Currency, -take)
	if err != nil {
		return nil, err
	}
	walletTx := newWalletTransaction(wallet, &in.WalletEntry, -take, balance, metadata)
	if err := p.repo.CreateTransaction(ctx, walletTx); err != nil {
		return nil, fmt.Errorf("failed to create wallet transaction: %w", err)
	}
	result.Transaction = walletTx

	var allocations []entities.WalletLotAllocation
	remaining := take
	for _, lot := range lots {
		if remaining == 0 {
			break
		}
		amount := min(lot.RemainingAmount, remaining)
		remaining -= amount
		if err := p.repo.ConsumeLot(ctx, lot.ID, amount); err != nil {
			return nil, err
		}
		allocations = append(allocations, entities.WalletLotAllocation{TransactionID: walletTx.ID, LotID: lot.ID, Amount: amount})
		result.Allocations = append(result.Allocations, WalletAllocation{LotID: lot.ID, Amount: amount, ExpiresAt: lot.ExpiresAt})
	}
	if err := p.repo.CreateAllocations(ctx, allocations); err != nil {
		return nil, fmt.Errorf("failed to create wallet lot allocations: %w", err)
	}
	return result, nil
}

// spendableLots returns the lots a debit may draw from, in the order it draws: the
// preferred lots first, then the unexpired lots in K9 order.
func (p *WalletProcessor) spendableLots(ctx context.Context, in *WalletDebitInput) ([]entities.WalletLot, error) {
	var lots []entities.WalletLot
	preferred := make(map[uuid.UUID]bool, len(in.PreferredLotIDs))

	if len(in.PreferredLotIDs) > 0 {
		found, err := p.repo.GetLotsByIDs(ctx, in.PreferredLotIDs)
		if err != nil {
			return nil, err
		}
		byID := make(map[uuid.UUID]entities.WalletLot, len(found))
		for _, lot := range found {
			byID[lot.ID] = lot
		}
		for _, id := range in.PreferredLotIDs {
			lot, ok := byID[id]
			if !ok || lot.CustomerID != in.CustomerID || lot.Currency != in.Currency {
				return nil, fmt.Errorf("%w: lot %s is not a %s lot of this customer", ErrWalletInvalidEntry, id, in.Currency)
			}
			if preferred[id] {
				continue
			}
			preferred[id] = true
			if lot.RemainingAmount > 0 {
				lots = append(lots, lot)
			}
		}
	}

	active, err := p.repo.ListActiveLots(ctx, in.CustomerID, in.Currency, p.now())
	if err != nil {
		return nil, err
	}
	for _, lot := range active {
		if !preferred[lot.ID] {
			lots = append(lots, lot)
		}
	}
	return lots, nil
}

// replay returns the first result for an idempotency key that has already been used,
// or nil when the key is new. It runs after the wallet lock, so a concurrent request
// with the same key has either committed its row or not started.
func (p *WalletProcessor) replay(ctx context.Context, in *WalletEntry, credit, compareAmount bool) (*WalletResult, error) {
	if in.IdempotencyKey == "" {
		return nil, nil
	}
	walletTx, err := p.repo.GetTransactionByIdempotencyKey(ctx, in.IdempotencyKey)
	if err != nil || walletTx == nil {
		return nil, err
	}

	sameDirection := (walletTx.Amount > 0) == credit
	sameAmount := !compareAmount || abs(walletTx.Amount) == in.Amount
	if walletTx.CustomerID != in.CustomerID || walletTx.Currency != in.Currency ||
		walletTx.Type != in.Type || !sameDirection || !sameAmount {
		return nil, ErrWalletIdempotencyConflict
	}

	result := &WalletResult{Transaction: walletTx, Replayed: true}
	if credit {
		result.Lots, err = p.repo.ListLotsBySourceTransaction(ctx, walletTx.ID)
		return result, err
	}

	// JSON numbers come back from JSONB as float64.
	switch shortfall := walletTx.Metadata["shortfall"].(type) {
	case float64:
		result.Shortfall = int64(shortfall)
	case int64:
		result.Shortfall = shortfall
	}
	allocations, err := p.repo.ListAllocationsByTransaction(ctx, walletTx.ID)
	if err != nil {
		return nil, err
	}
	ids := make([]uuid.UUID, 0, len(allocations))
	for _, a := range allocations {
		ids = append(ids, a.LotID)
	}
	lots, err := p.repo.GetLotsByIDs(ctx, ids)
	if err != nil {
		return nil, err
	}
	expiry := make(map[uuid.UUID]*time.Time, len(lots))
	for _, lot := range lots {
		expiry[lot.ID] = lot.ExpiresAt
	}
	for _, a := range allocations {
		result.Allocations = append(result.Allocations, WalletAllocation{LotID: a.LotID, Amount: a.Amount, ExpiresAt: expiry[a.LotID]})
	}
	return result, nil
}

func newWalletTransaction(wallet *entities.CustomerWallet, in *WalletEntry, amount, balance int64, extra entities.Metadata) *entities.WalletTransaction {
	metadata := entities.Metadata{}
	for k, v := range in.Metadata {
		metadata[k] = v
	}
	for k, v := range extra {
		metadata[k] = v
	}
	var key *string
	if in.IdempotencyKey != "" {
		k := in.IdempotencyKey
		key = &k
	}
	return &entities.WalletTransaction{
		ID:                     in.TransactionID,
		OrganizationID:         wallet.OrganizationID,
		CustomerID:             in.CustomerID,
		Currency:               in.Currency,
		Type:                   in.Type,
		Amount:                 amount,
		BalanceAfter:           balance,
		GroupID:                in.GroupID,
		ReferenceType:          in.ReferenceType,
		ReferenceID:            in.ReferenceID,
		CounterpartyCustomerID: in.CounterpartyCustomerID,
		ReversesTransactionID:  in.ReversesTransactionID,
		OutletID:               in.OutletID,
		CreatedByUser:          in.CreatedByUser,
		Reason:                 in.Reason,
		Description:            in.Description,
		Metadata:               metadata,
		IdempotencyKey:         key,
	}
}

// walletTypeRule is one row of §8.1.
type walletTypeRule struct {
	credit, debit  bool
	currency       string // empty: either currency
	referenceTypes []string
	needsOutlet    bool
	needsReverses  bool
	needsGroup     bool
	needsCounter   bool
	needsActor     bool
}

var walletTypeRules = map[string]walletTypeRule{
	constants.WalletTxTypeEarn:          {credit: true, referenceTypes: []string{constants.WalletRefTypeOrder}, needsOutlet: true},
	constants.WalletTxTypeEarnReversal:  {debit: true, referenceTypes: []string{constants.WalletRefTypeOrder}, needsOutlet: true, needsReverses: true},
	constants.WalletTxTypePayment:       {debit: true, currency: constants.WalletCurrencyPoint, referenceTypes: []string{constants.WalletRefTypePayment}, needsOutlet: true},
	constants.WalletTxTypePaymentRefund: {credit: true, currency: constants.WalletCurrencyPoint, referenceTypes: []string{constants.WalletRefTypePayment}, needsOutlet: true, needsReverses: true},
	constants.WalletTxTypeExchangeOut:   {debit: true, currency: constants.WalletCurrencyCoin, referenceTypes: []string{constants.WalletRefTypeWalletTx}, needsGroup: true},
	constants.WalletTxTypeExchangeIn:    {credit: true, currency: constants.WalletCurrencyPoint, referenceTypes: []string{constants.WalletRefTypeWalletTx}, needsGroup: true},
	constants.WalletTxTypeTransferOut:   {debit: true, referenceTypes: []string{constants.WalletRefTypeWalletTx}, needsGroup: true, needsCounter: true},
	constants.WalletTxTypeTransferIn:    {credit: true, referenceTypes: []string{constants.WalletRefTypeWalletTx}, needsGroup: true, needsCounter: true},
	constants.WalletTxTypeGameSpend:     {debit: true, currency: constants.WalletCurrencyCoin, referenceTypes: []string{constants.WalletRefTypeGamePlay}},
	constants.WalletTxTypeExpire:        {debit: true, referenceTypes: []string{constants.WalletRefTypeLot}},
	constants.WalletTxTypeAdjustment:    {credit: true, debit: true, referenceTypes: []string{constants.WalletRefTypeUser}, needsActor: true},
	constants.WalletTxTypeMigration:     {credit: true, referenceTypes: []string{constants.WalletRefTypeLegacyPoints, constants.WalletRefTypeLegacyTokens}},
	constants.WalletTxTypeRewardRedeem:  {debit: true, currency: constants.WalletCurrencyPoint, referenceTypes: []string{constants.WalletRefTypeRewardRedemption}},
}

func validateWalletEntry(in *WalletEntry, credit bool) error {
	invalid := func(format string, args ...any) error {
		return fmt.Errorf("%w: %s", ErrWalletInvalidEntry, fmt.Sprintf(format, args...))
	}

	rule, ok := walletTypeRules[in.Type]
	if !ok {
		return invalid("unknown type %q", in.Type)
	}
	if credit && !rule.credit {
		return invalid("%s cannot add to a balance", in.Type)
	}
	if !credit && !rule.debit {
		return invalid("%s cannot take from a balance", in.Type)
	}
	if in.CustomerID == uuid.Nil {
		return invalid("customer is required")
	}
	if !constants.IsValidWalletCurrency(in.Currency) {
		return invalid("unknown currency %q", in.Currency)
	}
	if rule.currency != "" && in.Currency != rule.currency {
		return invalid("%s must be in %s", in.Type, rule.currency)
	}
	if in.Amount <= 0 {
		return invalid("amount must be positive, got %d", in.Amount)
	}
	if !containsString(rule.referenceTypes, in.ReferenceType) {
		return invalid("%s must reference %s, got %q", in.Type, strings.Join(rule.referenceTypes, " or "), in.ReferenceType)
	}
	if in.ReferenceID == uuid.Nil {
		return invalid("reference id is required")
	}
	if strings.TrimSpace(in.Description) == "" {
		return invalid("description is required")
	}
	if rule.needsOutlet && isNilID(in.OutletID) {
		return invalid("%s requires an outlet", in.Type)
	}
	if rule.needsReverses && isNilID(in.ReversesTransactionID) {
		return invalid("%s requires the transaction it reverses", in.Type)
	}
	if rule.needsGroup && isNilID(in.GroupID) {
		return invalid("%s requires a group id", in.Type)
	}
	if rule.needsCounter {
		if isNilID(in.CounterpartyCustomerID) {
			return invalid("%s requires a counterparty", in.Type)
		}
		if *in.CounterpartyCustomerID == in.CustomerID {
			return invalid("%s cannot go to the same customer", in.Type)
		}
	}
	if rule.needsActor {
		if isNilID(in.CreatedByUser) {
			return invalid("%s requires the admin who made it", in.Type)
		}
		if in.Reason == nil || strings.TrimSpace(*in.Reason) == "" {
			return invalid("%s requires a reason", in.Type)
		}
	}
	return nil
}

func isNilID(id *uuid.UUID) bool {
	return id == nil || *id == uuid.Nil
}

func containsString(values []string, v string) bool {
	for _, value := range values {
		if value == v {
			return true
		}
	}
	return false
}

func abs(v int64) int64 {
	if v < 0 {
		return -v
	}
	return v
}