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 } // 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 }