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") ) // 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") // 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 } // 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, ).VALUES( o.orderID, o.accountID, o.platform, o.productID, o.amount.Minor(), string(o.amount.Currency()), string(o.origin), "pending", o.provider, now, now, ) 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 } // 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 orderRow{ orderID: o.OrderID, accountID: o.AccountID, productID: o.ProductID, expectedAmount: o.ExpectedAmount, currency: o.Currency, origin: o.Origin, status: o.Status, }, 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 } // 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 } // 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" }