Files
scrabble-game/backend/internal/payments/store_intake.go
T
Ilia Denisov 4cac09c9f3
CI / changes (pull_request) Successful in 3s
CI / unit (pull_request) Successful in 11s
CI / integration (pull_request) Successful in 21s
CI / ui (pull_request) Successful in 1m16s
CI / conformance (pull_request) Successful in 10s
CI / gate (pull_request) Successful in 0s
CI / deploy (pull_request) Successful in 1m55s
fix(payments): drop the notification sender-IP gate; re-check on its own cadence
A real test payment on the contour exposed both problems at once. YooKassa
delivered the notification five times; all five were rejected because the
backend saw the sender as 10.77.0.1 — the contour sits behind a tunnel and
cannot observe real client addresses, the same reason the IP bans in this
repository are prod-only. The chips were not lost (the reconcile sweep would
have credited them), but the primary path was dead and the customer was left
watching an unchanged balance.

The address check is removed rather than made conditional. It never was the
security boundary — the confirming GET /v3/payments/{id} is — and the one thing
it bought is already bought earlier and far more tightly: the order is resolved
from the notification's metadata *before* any provider call, so a notification
naming no known order costs a single indexed read and stops there. Guessing a
live order id means guessing a uuid. Against that, an address check adds nothing
and breaks every deployment that cannot see real client addresses, while turning
any future change to YooKassa's published ranges into a silent degradation.

The second problem was mine. The reconcile threshold was keyed off the order
lifetime, so a lost notification cost the customer the full 30-minute TTL before
the chips landed. Those are different questions: the lifetime governs how long a
customer may take to pay, the re-check governs how soon we notice a lost
callback. Split apart — `payments.ReconcileAfter`, one minute, swept on every
reaper tick. The bound D49 was chosen for survives: the calls one order can
cause are still its lifetime divided by the sweep interval, a handful, not an
open-ended poll. Worst case for a failed notification drops from ~30 minutes to
~5; an order the customer is still paying for is left alone.

Tests: the foreign-sender test is replaced by the two properties that now carry
the load — a notification naming an unknown order makes no provider call at all,
and a genuine notification is honoured whatever address it appears to come from.
Plus one pinning that a seconds-old order is not polled.

The shared bundle budget goes 31 -> 32 KB, with the reason recorded in the
script header: every user-visible string lands in that chunk and it had been
sitting 40 bytes under the cap.

Decisions D48 and D49 revised.
2026-07-28 12:00:03 +02:00

722 lines
28 KiB
Go

package payments
import (
"context"
"database/sql"
"encoding/json"
"errors"
"fmt"
"time"
"github.com/go-jet/jet/v2/postgres"
"github.com/go-jet/jet/v2/qrm"
"github.com/google/uuid"
"github.com/jackc/pgx/v5/pgconn"
"scrabble/backend/internal/postgres/jet/payments/model"
"scrabble/backend/internal/postgres/jet/payments/table"
)
// Intake errors surfaced by the order-flow and external-credit (fund) path.
var (
// ErrNotAPack means the product is not a fundable chip pack in the requested method: it
// carries no chips atom, or no price for that payment method.
ErrNotAPack = errors.New("payments: product is not a chip pack for this method")
// ErrOrderNotFound means no order matches the id (an unknown or forged callback reference).
ErrOrderNotFound = errors.New("payments: order not found")
// ErrAmountMismatch means the callback's paid amount or currency does not match the order's
// expected amount — the credit is refused (§9: verify amount after matching by order id).
ErrAmountMismatch = errors.New("payments: paid amount does not match the order")
// ErrOrderNotPaid means a refund targets an order that was never funded — there is nothing to
// reverse (guards against a spurious loss/abuse record on an unpaid order).
ErrOrderNotPaid = errors.New("payments: order is not paid")
)
// errAlreadyCredited is the internal sentinel that unwinds the fund transaction when the ledger's
// (provider, provider_payment_id) unique index rejects a duplicate callback. It is not surfaced:
// a replayed callback is a success that credits nothing.
var errAlreadyCredited = errors.New("payments: already credited")
// errAlreadyRefunded unwinds the refund transaction when the ledger idempotency index rejects a
// duplicate refund (same provider refund id). It is not surfaced: a replayed refund reverses nothing.
var errAlreadyRefunded = errors.New("payments: already refunded")
// packInfo is a chip pack resolved for an order: the product, the chips it funds and its price in
// the requested payment method's currency.
type packInfo struct {
productID uuid.UUID
title string
chips int
price Money
}
// packChips returns the quantity of the chips atom a product carries, or 0 if it has none (which
// marks it as a value, not a fundable pack).
func (s *Store) packChips(ctx context.Context, productID uuid.UUID) (int, error) {
var item model.ProductItem
err := postgres.SELECT(table.ProductItem.AllColumns).
FROM(table.ProductItem).
WHERE(table.ProductItem.ProductID.EQ(postgres.UUID(productID)).
AND(table.ProductItem.AtomType.EQ(postgres.String(atomChips)))).
LIMIT(1).
QueryContext(ctx, s.db, &item)
if errors.Is(err, qrm.ErrNoRows) {
return 0, nil
}
if err != nil {
return 0, fmt.Errorf("payments: load pack chips %s: %w", productID, err)
}
return int(item.Quantity), nil
}
// loadPackForOrder resolves an active chip pack for a new order in the given payment method: an
// active product carrying a chips atom and a price row for the method (its currency and amount are
// the order's expected amount). It rejects a missing/deactivated product (ErrProductNotFound) and
// a product that is not a pack for the method (ErrNotAPack).
func (s *Store) loadPackForOrder(ctx context.Context, productID uuid.UUID, method Source) (packInfo, error) {
var p model.Product
err := postgres.SELECT(table.Product.AllColumns).
FROM(table.Product).
WHERE(table.Product.ProductID.EQ(postgres.UUID(productID))).
LIMIT(1).
QueryContext(ctx, s.db, &p)
if errors.Is(err, qrm.ErrNoRows) || (err == nil && !p.Active) {
return packInfo{}, ErrProductNotFound
}
if err != nil {
return packInfo{}, fmt.Errorf("payments: load product %s: %w", productID, err)
}
chips, err := s.packChips(ctx, productID)
if err != nil {
return packInfo{}, err
}
if chips <= 0 {
return packInfo{}, ErrNotAPack
}
var price model.ProductPrice
err = postgres.SELECT(table.ProductPrice.AllColumns).
FROM(table.ProductPrice).
WHERE(table.ProductPrice.ProductID.EQ(postgres.UUID(productID)).
AND(table.ProductPrice.Method.EQ(postgres.String(string(method))))).
LIMIT(1).
QueryContext(ctx, s.db, &price)
if errors.Is(err, qrm.ErrNoRows) {
return packInfo{}, ErrNotAPack
}
if err != nil {
return packInfo{}, fmt.Errorf("payments: load pack price %s: %w", productID, err)
}
money, err := MoneyFromMinor(price.Amount, Currency(price.Currency))
if err != nil {
return packInfo{}, err
}
return packInfo{productID: productID, title: p.Title, chips: chips, price: money}, nil
}
// packForCredit resolves the chips and title of an ordered pack at credit time, ignoring the
// product's active flag: the money is real, so an order is honoured even if the pack was
// deactivated after it was placed (§9/D23).
func (s *Store) packForCredit(ctx context.Context, productID uuid.UUID) (chips int, title string, err error) {
var p model.Product
e := postgres.SELECT(table.Product.Title).
FROM(table.Product).
WHERE(table.Product.ProductID.EQ(postgres.UUID(productID))).
LIMIT(1).
QueryContext(ctx, s.db, &p)
if errors.Is(e, qrm.ErrNoRows) {
return 0, "", ErrProductNotFound
}
if e != nil {
return 0, "", fmt.Errorf("payments: load product %s: %w", productID, e)
}
chips, err = s.packChips(ctx, productID)
if err != nil {
return 0, "", err
}
if chips <= 0 {
return 0, "", ErrNotAPack
}
return chips, p.Title, nil
}
// newOrder is the intent a CreateOrder writes: a pending order for a pack, priced in the method's
// currency, tagged with the provider that will settle it.
type newOrder struct {
orderID uuid.UUID
accountID uuid.UUID
platform string
productID uuid.UUID
amount Money
origin Source
provider string
shop string
}
// createOrder inserts a pending order.
func (s *Store) createOrder(ctx context.Context, o newOrder, now time.Time) error {
stmt := table.Orders.INSERT(
table.Orders.OrderID, table.Orders.AccountID, table.Orders.Platform,
table.Orders.ProductID, table.Orders.ExpectedAmount, table.Orders.Currency,
table.Orders.Origin, table.Orders.Status, table.Orders.Provider,
table.Orders.CreatedAt, table.Orders.UpdatedAt, table.Orders.Shop,
).VALUES(
o.orderID, o.accountID, o.platform,
o.productID, o.amount.Minor(), string(o.amount.Currency()),
string(o.origin), "pending", o.provider,
now, now, o.shop,
)
if _, err := stmt.ExecContext(ctx, s.db); err != nil {
return fmt.Errorf("payments: create order: %w", err)
}
return nil
}
// orderRow is a stored order read back for the intake path.
type orderRow struct {
orderID uuid.UUID
accountID uuid.UUID
productID uuid.UUID
expectedAmount int64
currency string
origin string
status string
provider string
paymentID string
shop string
}
// orderByID reads an order, or ErrOrderNotFound.
func (s *Store) orderByID(ctx context.Context, orderID uuid.UUID) (orderRow, error) {
var o model.Orders
err := postgres.SELECT(table.Orders.AllColumns).
FROM(table.Orders).
WHERE(table.Orders.OrderID.EQ(postgres.UUID(orderID))).
LIMIT(1).
QueryContext(ctx, s.db, &o)
if errors.Is(err, qrm.ErrNoRows) {
return orderRow{}, ErrOrderNotFound
}
if err != nil {
return orderRow{}, fmt.Errorf("payments: load order %s: %w", orderID, err)
}
return newOrderRow(o), nil
}
// newOrderRow projects a stored order onto the intake's view of it, flattening the nullable provider
// columns to their empty strings.
func newOrderRow(o model.Orders) orderRow {
return orderRow{
orderID: o.OrderID,
accountID: o.AccountID,
productID: o.ProductID,
expectedAmount: o.ExpectedAmount,
currency: o.Currency,
origin: o.Origin,
status: o.Status,
provider: derefString(o.Provider),
paymentID: derefString(o.ProviderPaymentID),
shop: o.Shop,
}
}
// derefString reads a nullable text column as a plain string, treating NULL as empty.
func derefString(p *string) string {
if p == nil {
return ""
}
return *p
}
// orderByProviderPayment reads the order a provider's payment id belongs to, or ErrOrderNotFound.
// It is how a provider event that names only its own payment — a refund notification, say — is
// resolved back to an order.
func (s *Store) orderByProviderPayment(ctx context.Context, provider, providerPaymentID string) (orderRow, error) {
if provider == "" || providerPaymentID == "" {
return orderRow{}, ErrOrderNotFound
}
var o model.Orders
err := postgres.SELECT(table.Orders.AllColumns).
FROM(table.Orders).
WHERE(table.Orders.Provider.EQ(postgres.String(provider)).
AND(table.Orders.ProviderPaymentID.EQ(postgres.String(providerPaymentID)))).
LIMIT(1).
QueryContext(ctx, s.db, &o)
if errors.Is(err, qrm.ErrNoRows) {
return orderRow{}, ErrOrderNotFound
}
if err != nil {
return orderRow{}, fmt.Errorf("payments: load order by %s payment %s: %w", provider, providerPaymentID, err)
}
return newOrderRow(o), nil
}
// attachProviderPayment records the provider's own payment id on a pending order, as soon as the
// provider mints it. It is what later lets an unattended order be re-checked against the provider
// and a refund address the right payment; fund overwrites it with the same value when the callback
// lands. It never changes the order status.
func (s *Store) attachProviderPayment(ctx context.Context, orderID uuid.UUID, provider, providerPaymentID string, now time.Time) error {
_, err := table.Orders.
UPDATE(table.Orders.Provider, table.Orders.ProviderPaymentID, table.Orders.UpdatedAt).
SET(postgres.String(provider), postgres.String(providerPaymentID), postgres.TimestampzT(now)).
WHERE(table.Orders.OrderID.EQ(postgres.UUID(orderID))).
ExecContext(ctx, s.db)
if err != nil {
return fmt.Errorf("payments: attach provider payment to order %s: %w", orderID, err)
}
return nil
}
// pendingForReconcile reads the pending orders older than ageSeconds that carry a provider payment
// id — the ones whose real outcome is still unknown to us because no callback ever arrived. The
// caller asks the provider what happened. Orders with no provider payment id are skipped: the
// customer never got as far as a payment.
func (s *Store) pendingForReconcile(ctx context.Context, ageSeconds int, now time.Time, limit int) ([]orderRow, error) {
cutoff := now.Add(-time.Duration(ageSeconds) * time.Second)
var rows []model.Orders
err := postgres.SELECT(table.Orders.AllColumns).
FROM(table.Orders).
WHERE(table.Orders.Status.EQ(postgres.String("pending")).
AND(table.Orders.CreatedAt.LT(postgres.TimestampzT(cutoff))).
AND(table.Orders.ProviderPaymentID.IS_NOT_NULL()).
AND(table.Orders.ProviderPaymentID.NOT_EQ(postgres.String("")))).
ORDER_BY(table.Orders.CreatedAt.ASC()).
LIMIT(int64(limit)).
QueryContext(ctx, s.db, &rows)
if err != nil && !errors.Is(err, qrm.ErrNoRows) {
return nil, fmt.Errorf("payments: load orders for reconcile: %w", err)
}
out := make([]orderRow, 0, len(rows))
for _, r := range rows {
out = append(out, newOrderRow(r))
}
return out, nil
}
// FundOutcome reports the result of an intake credit: whose balance, which segment and how many
// chips were credited, and whether the callback was a duplicate that credited nothing.
type FundOutcome struct {
AccountID uuid.UUID
Source Source
Chips int
AlreadyCredited bool
}
// fund credits a paid order exactly once: it matches the order, verifies the amount, then in one
// transaction appends a fund ledger row (idempotent on the (provider, provider_payment_id) unique
// index), credits the funded segment's balance and marks the order paid. A duplicate callback is
// rejected by the unique index and returns AlreadyCredited with no error and no second credit. A
// valid callback is honoured even on an expired order (§9/D23). The read cache is invalidated after
// the commit, since the credit runs outside any request the payments package owns.
func (s *Store) fund(ctx context.Context, orderID uuid.UUID, provider, providerPaymentID string, paid Money, now time.Time) (FundOutcome, error) {
ord, err := s.orderByID(ctx, orderID)
if err != nil {
return FundOutcome{}, err
}
if paid.Currency() != Currency(ord.currency) || paid.Minor() != ord.expectedAmount {
return FundOutcome{}, ErrAmountMismatch
}
chips, title, err := s.packForCredit(ctx, ord.productID)
if err != nil {
return FundOutcome{}, err
}
snapshot, err := marshalFundSnapshot(ord.productID, title, chips, paid)
if err != nil {
return FundOutcome{}, err
}
src := Source(ord.origin)
outcome := FundOutcome{AccountID: ord.accountID, Source: src, Chips: chips}
pv, pp := provider, providerPaymentID
productID := ord.productID
err = withTx(ctx, s.db, func(tx *sql.Tx) error {
if e := insertLedgerTx(ctx, tx, ord.accountID, "fund", &src, &src, chips, &productID, &orderID, &pv, &pp, snapshot, now); e != nil {
if isUniqueViolation(e) {
outcome.AlreadyCredited = true
return errAlreadyCredited
}
return e
}
if _, e := tx.ExecContext(ctx,
`INSERT INTO payments.balances (account_id, source, chips, updated_at)
VALUES ($1, $2, $3, now())
ON CONFLICT (account_id, source) DO UPDATE
SET chips = payments.balances.chips + EXCLUDED.chips, updated_at = now()`,
ord.accountID, string(src), chips); e != nil {
return fmt.Errorf("payments: credit balance %s: %w", src, e)
}
if _, e := tx.ExecContext(ctx,
`UPDATE payments.orders SET status = 'paid', provider = $2, provider_payment_id = $3, updated_at = now()
WHERE order_id = $1`,
orderID, provider, providerPaymentID); e != nil {
return fmt.Errorf("payments: mark order paid: %w", e)
}
return nil
})
if err != nil {
if errors.Is(err, errAlreadyCredited) {
return outcome, nil
}
return FundOutcome{}, err
}
s.cache.invalidate(ord.accountID)
return outcome, nil
}
// RefundOutcome reports a refund's result: whose funded segment was reversed, the chips actually
// clawed back (floored at 0), the unrecoverable remainder (a loss, when the chips were already
// spent), and whether the refund was a duplicate that reversed nothing.
type RefundOutcome struct {
AccountID uuid.UUID
Source Source
Revoked int
Loss int
AlreadyRefunded bool
}
// refund reverses a paid order's credit best-effort, exactly once. It matches the order (which must
// be paid), verifies the refunded amount, then in one transaction appends a refund ledger row
// (idempotent on the (provider, provider_payment_id) index — the refund id is distinct from the
// fund's payment id, so the two rows coexist), revokes the funded chips floored at 0 (never
// negative, D27/balances_chips_chk) and, when chips were already spent, records the unrecoverable
// remainder as a per-account loss and flips the abuse flag. A duplicate refund returns
// AlreadyRefunded with no second reversal. The ledger row's chipsDelta is what is actually
// reclaimed; the full reversal (money, original chips, loss) rides in its snapshot for the report.
func (s *Store) refund(ctx context.Context, orderID uuid.UUID, provider, providerRefundID string, refunded Money, now time.Time) (RefundOutcome, error) {
ord, err := s.orderByID(ctx, orderID)
if err != nil {
return RefundOutcome{}, err
}
if ord.status != "paid" {
return RefundOutcome{}, ErrOrderNotPaid
}
if refunded.Currency() != Currency(ord.currency) || refunded.Minor() != ord.expectedAmount {
return RefundOutcome{}, ErrAmountMismatch
}
chips, title, err := s.packForCredit(ctx, ord.productID)
if err != nil {
return RefundOutcome{}, err
}
src := Source(ord.origin)
outcome := RefundOutcome{AccountID: ord.accountID, Source: src}
pv, pr := provider, providerRefundID
productID := ord.productID
oid := orderID
err = withTx(ctx, s.db, func(tx *sql.Tx) error {
// Lock the funded segment and read what is left; a spent balance floors the reversal at 0.
var avail int
e := tx.QueryRowContext(ctx,
`SELECT chips FROM payments.balances WHERE account_id = $1 AND source = $2 FOR UPDATE`,
ord.accountID, string(src)).Scan(&avail)
switch {
case errors.Is(e, sql.ErrNoRows):
avail = 0
case e != nil:
return fmt.Errorf("payments: read balance for refund: %w", e)
}
revoked := min(chips, avail)
loss := chips - revoked
outcome.Revoked, outcome.Loss = revoked, loss
snapshot, e := marshalRefundSnapshot(ord.productID, title, chips, revoked, loss, refunded, providerRefundID)
if e != nil {
return e
}
if e := insertLedgerTx(ctx, tx, ord.accountID, "refund", &src, &src, -revoked, &productID, &oid, &pv, &pr, snapshot, now); e != nil {
if isUniqueViolation(e) {
outcome.AlreadyRefunded = true
return errAlreadyRefunded
}
return e
}
if revoked > 0 {
if _, e := tx.ExecContext(ctx,
`UPDATE payments.balances SET chips = chips - $3, updated_at = now()
WHERE account_id = $1 AND source = $2`,
ord.accountID, string(src), revoked); e != nil {
return fmt.Errorf("payments: revoke chips %s: %w", src, e)
}
}
if loss > 0 {
if _, e := tx.ExecContext(ctx,
`INSERT INTO payments.account_risk (account_id, abuse, loss_chips, updated_at)
VALUES ($1, true, $2, now())
ON CONFLICT (account_id) DO UPDATE
SET abuse = true, loss_chips = payments.account_risk.loss_chips + EXCLUDED.loss_chips, updated_at = now()`,
ord.accountID, int64(loss)); e != nil {
return fmt.Errorf("payments: record refund loss: %w", e)
}
}
return nil
})
if err != nil {
if errors.Is(err, errAlreadyRefunded) {
return outcome, nil
}
return RefundOutcome{}, err
}
s.cache.invalidate(ord.accountID)
return outcome, nil
}
// RewardOutcome reports a rewarded-video credit: the chips credited (0 when rewarded is unconfigured
// or the daily cap is reached), whether the daily cap blocked it, and whether it was a duplicate view
// (same client nonce) that credited nothing more.
type RewardOutcome struct {
AccountID uuid.UUID
Chips int
Capped bool
AlreadyCredited bool
}
// interstitialCooldowns reads the post-move interstitial-ad cooldowns (seconds): the global
// per-user cooldown, the longer vs_ai one, and the independent hint-triggered one. The client mirrors
// them and self-gates (E6/D30).
func (s *Store) interstitialCooldowns(ctx context.Context) (global, vsAi, hint int, err error) {
var cfg model.Config
if e := postgres.SELECT(table.Config.CooldownGlobalSeconds, table.Config.CooldownVsAiSeconds, table.Config.CooldownHintSeconds).
FROM(table.Config).
LIMIT(1).
QueryContext(ctx, s.db, &cfg); e != nil {
return 0, 0, 0, fmt.Errorf("payments: read interstitial cooldowns: %w", e)
}
return int(cfg.CooldownGlobalSeconds), int(cfg.CooldownVsAiSeconds), int(cfg.CooldownHintSeconds), nil
}
// rewardConfig reads the rewarded payout (chips per view) and the per-day and per-hour caps. The
// caps are both anti-abuse (bounding a forger's free chips) and an economic conversion lever (free
// rewarded chips are limited so a player who wants more buys) — tuned in the admin.
func (s *Store) rewardConfig(ctx context.Context) (payout, dailyCap, hourlyCap int, err error) {
var cfg model.Config
if e := postgres.SELECT(table.Config.RewardedPayoutChips, table.Config.RewardDailyCap, table.Config.RewardHourlyCap).
FROM(table.Config).
LIMIT(1).
QueryContext(ctx, s.db, &cfg); e != nil {
return 0, 0, 0, fmt.Errorf("payments: read reward config: %w", e)
}
return int(cfg.RewardedPayoutChips), int(cfg.RewardDailyCap), int(cfg.RewardHourlyCap), nil
}
// setRewardConfig updates the rewarded payout and the per-day and per-hour caps on the singleton
// config row (no WHERE — the table holds exactly one row). Non-negativity is validated by the caller
// and also enforced by the config CHECK constraints.
func (s *Store) setRewardConfig(ctx context.Context, payout, dailyCap, hourlyCap int) error {
if _, err := s.db.ExecContext(ctx,
`UPDATE payments.config SET rewarded_payout_chips=$1, reward_daily_cap=$2, reward_hourly_cap=$3`,
payout, dailyCap, hourlyCap); err != nil {
return fmt.Errorf("payments: set reward config: %w", err)
}
return nil
}
// creditReward credits a rewarded-video view's chips to the funded segment, client-attested (VK Mini
// App ads expose no server verify). It reads the payout and daily cap from config: a 0 payout
// (unconfigured) or a reached cap credits nothing. It is idempotent on the client nonce (dedup on the
// (provider, provider_payment_id) index), so a retried view credits once, and order-less (a free
// credit, no order). The cap counts today's rewarded credits for this network (UTC day); a rare
// concurrent race may allow cap+1, which the per-user rate limiter bounds and the cap tolerates.
func (s *Store) creditReward(ctx context.Context, accountID uuid.UUID, source Source, provider, nonce string, now time.Time) (RewardOutcome, error) {
payout, dailyCap, hourlyCap, err := s.rewardConfig(ctx)
if err != nil {
return RewardOutcome{}, err
}
outcome := RewardOutcome{AccountID: accountID}
if payout <= 0 {
return outcome, nil // rewarded not configured (0 payout) — inert until the owner sets it
}
// Count this network's rewarded credits in the last day and last hour (one scan over the last
// 25 h covers both windows); either cap reached blocks the credit. A rare concurrent race may
// allow cap+1, which the per-user rate limiter bounds and the anti-abuse cap tolerates.
var today, lastHour int
if e := s.db.QueryRowContext(ctx,
`SELECT count(*) FILTER (WHERE created_at >= date_trunc('day', now())),
count(*) FILTER (WHERE created_at >= now() - interval '1 hour')
FROM payments.ledger
WHERE account_id = $1 AND kind = 'fund' AND provider = $2 AND created_at >= now() - interval '25 hours'`,
accountID, provider).Scan(&today, &lastHour); e != nil {
return RewardOutcome{}, fmt.Errorf("payments: count rewarded views: %w", e)
}
if today >= dailyCap || lastHour >= hourlyCap {
outcome.Capped = true
return outcome, nil
}
snapshot, err := marshalRewardSnapshot(payout)
if err != nil {
return RewardOutcome{}, err
}
src := source
pv, pp := provider, nonce
err = withTx(ctx, s.db, func(tx *sql.Tx) error {
if e := insertLedgerTx(ctx, tx, accountID, "fund", &src, &src, payout, nil, nil, &pv, &pp, snapshot, now); e != nil {
if isUniqueViolation(e) {
outcome.AlreadyCredited = true
return errAlreadyCredited
}
return e
}
if _, e := tx.ExecContext(ctx,
`INSERT INTO payments.balances (account_id, source, chips, updated_at)
VALUES ($1, $2, $3, now())
ON CONFLICT (account_id, source) DO UPDATE
SET chips = payments.balances.chips + EXCLUDED.chips, updated_at = now()`,
accountID, string(src), payout); e != nil {
return fmt.Errorf("payments: credit rewarded balance: %w", e)
}
return nil
})
if err != nil {
if errors.Is(err, errAlreadyCredited) {
return outcome, nil
}
return RewardOutcome{}, err
}
outcome.Chips = payout
s.cache.invalidate(accountID)
return outcome, nil
}
// insertPaymentEvent appends an undispatched lifecycle event (succeeded/failed/refunded) for the
// dispatcher to deliver. orderID and payload (a jsonb detail blob) are optional.
func (s *Store) insertPaymentEvent(ctx context.Context, accountID uuid.UUID, orderID *uuid.UUID, eventType string, payload []byte, now time.Time) error {
id, err := uuid.NewV7()
if err != nil {
return fmt.Errorf("payments: event id: %w", err)
}
var pl any = postgres.NULL
if payload != nil {
pl = string(payload)
}
stmt := table.PaymentEvents.INSERT(
table.PaymentEvents.EventID, table.PaymentEvents.AccountID, table.PaymentEvents.OrderID,
table.PaymentEvents.Type, table.PaymentEvents.Payload, table.PaymentEvents.CreatedAt,
).VALUES(id, accountID, uuidOrNull(orderID), eventType, pl, now)
if _, err := stmt.ExecContext(ctx, s.db); err != nil {
return fmt.Errorf("payments: insert %s event: %w", eventType, err)
}
return nil
}
// PaymentEvent is an undispatched lifecycle event the dispatcher delivers to an account.
type PaymentEvent struct {
EventID uuid.UUID
AccountID uuid.UUID
Type string
}
// undispatchedEvents reads up to limit payment events not yet delivered, oldest first.
func (s *Store) undispatchedEvents(ctx context.Context, limit int) ([]PaymentEvent, error) {
rows, err := s.db.QueryContext(ctx,
`SELECT event_id, account_id, type FROM payments.payment_events
WHERE dispatched_at IS NULL ORDER BY created_at LIMIT $1`, limit)
if err != nil {
return nil, fmt.Errorf("payments: read undispatched events: %w", err)
}
defer rows.Close()
var out []PaymentEvent
for rows.Next() {
var e PaymentEvent
if err := rows.Scan(&e.EventID, &e.AccountID, &e.Type); err != nil {
return nil, fmt.Errorf("payments: scan event: %w", err)
}
out = append(out, e)
}
return out, rows.Err()
}
// markEventDispatched stamps an event as delivered so it is not re-sent.
func (s *Store) markEventDispatched(ctx context.Context, eventID uuid.UUID, now time.Time) error {
if _, err := s.db.ExecContext(ctx,
`UPDATE payments.payment_events SET dispatched_at = $2 WHERE event_id = $1`, eventID, now); err != nil {
return fmt.Errorf("payments: mark event dispatched: %w", err)
}
return nil
}
// orderTTL reads the configured pending-order lifetime in whole seconds.
func (s *Store) orderTTL(ctx context.Context) (int, error) {
var cfg model.Config
if err := postgres.SELECT(table.Config.OrderTTLSeconds).
FROM(table.Config).
LIMIT(1).
QueryContext(ctx, s.db, &cfg); err != nil {
return 0, fmt.Errorf("payments: read order ttl: %w", err)
}
return int(cfg.OrderTTLSeconds), nil
}
// expirePending marks every pending order older than ttlSeconds as expired, returning how many.
// Expiry is cosmetic DB hygiene: a later valid callback still credits an expired order (§9/D23).
func (s *Store) expirePending(ctx context.Context, ttlSeconds int, now time.Time) (int, error) {
cutoff := now.Add(-time.Duration(ttlSeconds) * time.Second)
res, err := table.Orders.
UPDATE(table.Orders.Status, table.Orders.UpdatedAt).
SET(postgres.String("expired"), postgres.TimestampzT(now)).
WHERE(table.Orders.Status.EQ(postgres.String("pending")).
AND(table.Orders.CreatedAt.LT(postgres.TimestampzT(cutoff)))).
ExecContext(ctx, s.db)
if err != nil {
return 0, fmt.Errorf("payments: expire pending orders: %w", err)
}
n, _ := res.RowsAffected()
return int(n), nil
}
// marshalFundSnapshot records what a fund credited (the pack, chips and paid amount) on the ledger
// row, so history stays independent of later catalog edits (§7/D34).
func marshalFundSnapshot(productID uuid.UUID, title string, chips int, paid Money) ([]byte, error) {
b, err := json.Marshal(struct {
ProductID string `json:"product_id"`
Title string `json:"title,omitempty"`
Chips int `json:"chips"`
Amount int64 `json:"amount_minor"`
Currency string `json:"currency"`
}{productID.String(), title, chips, paid.Minor(), string(paid.Currency())})
if err != nil {
return nil, fmt.Errorf("payments: marshal fund snapshot: %w", err)
}
return b, nil
}
// marshalRefundSnapshot records the full reversal on the refund ledger row: the pack, the original
// funded chips, how many were actually reclaimed, the unrecoverable loss (already spent), the money
// refunded and the provider refund id — so the ledger stays reconcilable against the balance
// (chipsDelta = revoked) while the report still sees the whole reversal (§7/D27/D34).
func marshalRefundSnapshot(productID uuid.UUID, title string, chips, revoked, loss int, refunded Money, refundID string) ([]byte, error) {
b, err := json.Marshal(struct {
ProductID string `json:"product_id"`
Title string `json:"title,omitempty"`
Chips int `json:"chips"`
Revoked int `json:"revoked"`
Loss int `json:"loss"`
Amount int64 `json:"amount_minor"`
Currency string `json:"currency"`
RefundID string `json:"refund_id"`
}{productID.String(), title, chips, revoked, loss, refunded.Minor(), string(refunded.Currency()), refundID})
if err != nil {
return nil, fmt.Errorf("payments: marshal refund snapshot: %w", err)
}
return b, nil
}
// marshalRewardSnapshot records a rewarded-video credit on its ledger row: the marker distinguishing
// it from a paid fund, and the chips granted — so the report separates ad-earned chips from purchases.
func marshalRewardSnapshot(chips int) ([]byte, error) {
b, err := json.Marshal(struct {
Reward bool `json:"reward"`
Chips int `json:"chips"`
}{true, chips})
if err != nil {
return nil, fmt.Errorf("payments: marshal reward snapshot: %w", err)
}
return b, nil
}
// isUniqueViolation reports whether err is a PostgreSQL unique-constraint violation (SQLSTATE
// 23505) — here, a duplicate provider callback hitting the ledger idempotency index.
func isUniqueViolation(err error) bool {
var pgErr *pgconn.PgError
return errors.As(err, &pgErr) && pgErr.Code == "23505"
}