scrabble-game/backend/internal/payments/store_intake.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
}

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

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

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

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