internal/store/postgres/donations.go

Rama main · réplica de lectura del repositorio original.

package postgres

import (
	"context"
	"errors"
	"fmt"

	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgtype"

	"telesrv/internal/domain"
	"telesrv/internal/store/postgres/sqlcgen"
)

type DonationStore struct {
	db sqlcgen.DBTX
}

func NewDonationStore(db sqlcgen.DBTX) *DonationStore {
	return &DonationStore{db: db}
}

func (s *DonationStore) WalletSeed(ctx context.Context) ([]byte, []byte, bool, error) {
	var seed, nonce []byte
	err := s.db.QueryRow(ctx, `SELECT encrypted_seed, nonce FROM donation_wallet WHERE id = 1`).Scan(&seed, &nonce)
	if errors.Is(err, pgx.ErrNoRows) {
		return nil, nil, false, nil
	}
	if err != nil {
		return nil, nil, false, fmt.Errorf("load donation wallet seed: %w", err)
	}
	return seed, nonce, true, nil
}

func (s *DonationStore) CreateWalletSeed(ctx context.Context, encryptedSeed, nonce []byte) error {
	_, err := s.db.Exec(ctx, `INSERT INTO donation_wallet (id, encrypted_seed, nonce) VALUES (1, $1, $2)`, encryptedSeed, nonce)
	if err != nil {
		if isUniqueViolation(err) {
			return domain.ErrDonationWalletAlreadyExists
		}
		return fmt.Errorf("create donation wallet seed: %w", err)
	}
	return nil
}

func (s *DonationStore) DonationAddressForUser(ctx context.Context, userID int64) (domain.DonationAddress, bool, error) {
	var out domain.DonationAddress
	err := s.db.QueryRow(ctx, `SELECT user_id, derivation_index, address, created_at
FROM donation_addresses WHERE user_id = $1`, userID).
		Scan(&out.UserID, &out.DerivationIndex, &out.Address, &out.CreatedAt)
	if errors.Is(err, pgx.ErrNoRows) {
		return domain.DonationAddress{}, false, nil
	}
	if err != nil {
		return domain.DonationAddress{}, false, fmt.Errorf("load donation address: %w", err)
	}
	return out, true, nil
}

func (s *DonationStore) DonationUserByAddress(ctx context.Context, address string) (int64, bool, error) {
	var userID int64
	err := s.db.QueryRow(ctx, `SELECT user_id FROM donation_addresses WHERE address = $1`, address).Scan(&userID)
	if errors.Is(err, pgx.ErrNoRows) {
		return 0, false, nil
	}
	if err != nil {
		return 0, false, fmt.Errorf("look up donation address owner: %w", err)
	}
	return userID, true, nil
}

func (s *DonationStore) NextDonationAddressIndex(ctx context.Context) (int64, error) {
	var index int64
	if err := s.db.QueryRow(ctx, `SELECT nextval('donation_address_index_seq')`).Scan(&index); err != nil {
		return 0, fmt.Errorf("reserve donation address index: %w", err)
	}
	return index, nil
}

func (s *DonationStore) InsertDonationAddress(ctx context.Context, userID, index int64, address string) (domain.DonationAddress, error) {
	var out domain.DonationAddress
	err := s.db.QueryRow(ctx, `
INSERT INTO donation_addresses (user_id, derivation_index, address) VALUES ($1, $2, $3)
ON CONFLICT (user_id) DO UPDATE SET user_id = donation_addresses.user_id
RETURNING user_id, derivation_index, address, created_at`, userID, index, address).
		Scan(&out.UserID, &out.DerivationIndex, &out.Address, &out.CreatedAt)
	if err != nil {
		return domain.DonationAddress{}, fmt.Errorf("insert donation address: %w", err)
	}
	return out, nil
}

func (s *DonationStore) EnabledDonationChains(ctx context.Context) ([]domain.DonationChain, error) {
	rows, err := s.db.Query(ctx, `SELECT chain_key, name, chain_id, rpc_url, ws_url, native_symbol, native_decimals,
confirmations_required, price_feed_address, manual_usd_rate_micros, enabled
FROM donation_chains WHERE enabled ORDER BY chain_key`)
	if err != nil {
		return nil, fmt.Errorf("list enabled donation chains: %w", err)
	}
	defer rows.Close()
	return scanDonationChains(rows)
}

func (s *DonationStore) DonationChain(ctx context.Context, chainKey string) (domain.DonationChain, bool, error) {
	var out domain.DonationChain
	err := s.db.QueryRow(ctx, `SELECT chain_key, name, chain_id, rpc_url, ws_url, native_symbol, native_decimals,
confirmations_required, price_feed_address, manual_usd_rate_micros, enabled
FROM donation_chains WHERE chain_key = $1`, chainKey).Scan(
		&out.Key, &out.Name, &out.ChainID, &out.RPCURL, &out.WSURL, &out.NativeSymbol, &out.NativeDecimals,
		&out.ConfirmationsRequired, &out.PriceFeedAddress, &out.ManualUSDRateMicros, &out.Enabled)
	if errors.Is(err, pgx.ErrNoRows) {
		return domain.DonationChain{}, false, nil
	}
	if err != nil {
		return domain.DonationChain{}, false, fmt.Errorf("load donation chain: %w", err)
	}
	return out, true, nil
}

func (s *DonationStore) UpdateDonationChainConfig(ctx context.Context, upd domain.DonationChainConfigUpdate) (domain.DonationChain, error) {
	var out domain.DonationChain
	err := s.db.QueryRow(ctx, `UPDATE donation_chains SET
    rpc_url = $2, ws_url = $3, confirmations_required = $4, price_feed_address = $5,
    manual_usd_rate_micros = $6, enabled = $7
WHERE chain_key = $1
RETURNING chain_key, name, chain_id, rpc_url, ws_url, native_symbol, native_decimals,
    confirmations_required, price_feed_address, manual_usd_rate_micros, enabled`,
		upd.ChainKey, upd.RPCURL, upd.WSURL, upd.ConfirmationsRequired, upd.PriceFeedAddress,
		upd.ManualUSDRateMicros, upd.Enabled).Scan(
		&out.Key, &out.Name, &out.ChainID, &out.RPCURL, &out.WSURL, &out.NativeSymbol, &out.NativeDecimals,
		&out.ConfirmationsRequired, &out.PriceFeedAddress, &out.ManualUSDRateMicros, &out.Enabled)
	if errors.Is(err, pgx.ErrNoRows) {
		return domain.DonationChain{}, domain.ErrDonationChainNotFound
	}
	if err != nil {
		return domain.DonationChain{}, fmt.Errorf("update donation chain config: %w", err)
	}
	return out, nil
}

func scanDonationChains(rows pgx.Rows) ([]domain.DonationChain, error) {
	out := make([]domain.DonationChain, 0)
	for rows.Next() {
		var c domain.DonationChain
		if err := rows.Scan(&c.Key, &c.Name, &c.ChainID, &c.RPCURL, &c.WSURL, &c.NativeSymbol, &c.NativeDecimals,
			&c.ConfirmationsRequired, &c.PriceFeedAddress, &c.ManualUSDRateMicros, &c.Enabled); err != nil {
			return nil, fmt.Errorf("scan donation chain: %w", err)
		}
		out = append(out, c)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate donation chains: %w", err)
	}
	return out, nil
}

func (s *DonationStore) DonationTokens(ctx context.Context, chainKey string) ([]domain.DonationToken, error) {
	rows, err := s.db.Query(ctx, `SELECT chain_key, symbol, contract_address, decimals
FROM donation_tokens WHERE chain_key = $1 ORDER BY symbol`, chainKey)
	if err != nil {
		return nil, fmt.Errorf("list donation tokens: %w", err)
	}
	defer rows.Close()
	out := make([]domain.DonationToken, 0)
	for rows.Next() {
		var t domain.DonationToken
		if err := rows.Scan(&t.ChainKey, &t.Symbol, &t.ContractAddress, &t.Decimals); err != nil {
			return nil, fmt.Errorf("scan donation token: %w", err)
		}
		out = append(out, t)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate donation tokens: %w", err)
	}
	return out, nil
}

func (s *DonationStore) DonationChainCursor(ctx context.Context, chainKey string) (int64, error) {
	var block int64
	err := s.db.QueryRow(ctx, `SELECT last_scanned_block FROM donation_chain_cursors WHERE chain_key = $1`, chainKey).Scan(&block)
	if errors.Is(err, pgx.ErrNoRows) {
		return 0, nil
	}
	if err != nil {
		return 0, fmt.Errorf("load donation chain cursor: %w", err)
	}
	return block, nil
}

func (s *DonationStore) SetDonationChainCursor(ctx context.Context, chainKey string, block int64) error {
	_, err := s.db.Exec(ctx, `
INSERT INTO donation_chain_cursors (chain_key, last_scanned_block, updated_at) VALUES ($1, $2, now())
ON CONFLICT (chain_key) DO UPDATE SET last_scanned_block = EXCLUDED.last_scanned_block, updated_at = now()`,
		chainKey, block)
	if err != nil {
		return fmt.Errorf("set donation chain cursor: %w", err)
	}
	return nil
}

// donationDepositScanner is the common surface of pgx.Row and pgx.Rows, so
// one scan routine serves every deposit query below regardless of whether
// it reads one row or iterates many.
type donationDepositScanner interface {
	Scan(dest ...any) error
}

// donationDepositColumns is the SELECT list every deposit query below uses,
// in the exact order scanDonationDeposit expects.
const donationDepositColumns = `id, user_id, chain_key, token_symbol, tx_hash, log_index, block_number, amount_raw::text,
       usd_value_micros, stars_credited, status, confirmations, detected_at, credited_at`

// scanDonationDeposit reads one donationDepositColumns row. credited_at is
// nullable (a deposit that hasn't been credited yet), so it scans through
// pgtype.Timestamptz rather than directly into time.Time.
func scanDonationDeposit(row donationDepositScanner) (domain.DonationDeposit, error) {
	var d domain.DonationDeposit
	var status string
	var creditedAt pgtype.Timestamptz
	if err := row.Scan(&d.ID, &d.UserID, &d.ChainKey, &d.TokenSymbol, &d.TxHash, &d.LogIndex, &d.BlockNumber,
		&d.AmountRaw, &d.USDValueMicros, &d.StarsCredited, &status, &d.Confirmations, &d.DetectedAt, &creditedAt); err != nil {
		return domain.DonationDeposit{}, err
	}
	d.Status = domain.DonationDepositStatus(status)
	if creditedAt.Valid {
		d.CreditedAt = creditedAt.Time
	}
	return d, nil
}

func scanDonationDeposits(rows pgx.Rows) ([]domain.DonationDeposit, error) {
	out := make([]domain.DonationDeposit, 0)
	for rows.Next() {
		d, err := scanDonationDeposit(rows)
		if err != nil {
			return nil, fmt.Errorf("scan donation deposit: %w", err)
		}
		out = append(out, d)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate donation deposits: %w", err)
	}
	return out, nil
}

func (s *DonationStore) RecordDonationDeposit(ctx context.Context, deposit domain.DonationDeposit) (domain.DonationDeposit, bool, error) {
	row := s.db.QueryRow(ctx, `
WITH inserted AS (
    INSERT INTO donation_deposits
        (user_id, chain_key, token_symbol, tx_hash, log_index, block_number, amount_raw, status, confirmations, detected_at)
    VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, now())
    ON CONFLICT (chain_key, tx_hash, log_index) DO NOTHING
    RETURNING *
)
SELECT `+donationDepositColumns+`, true AS created
FROM inserted
UNION ALL
SELECT `+donationDepositColumns+`, false AS created
FROM donation_deposits WHERE chain_key = $2 AND tx_hash = $4 AND log_index = $5
AND NOT EXISTS (SELECT 1 FROM inserted)
LIMIT 1`,
		deposit.UserID, deposit.ChainKey, deposit.TokenSymbol, deposit.TxHash, deposit.LogIndex,
		deposit.BlockNumber, deposit.AmountRaw, string(deposit.Status), deposit.Confirmations)
	var d domain.DonationDeposit
	var status string
	var creditedAt pgtype.Timestamptz
	var created bool
	if err := row.Scan(&d.ID, &d.UserID, &d.ChainKey, &d.TokenSymbol, &d.TxHash, &d.LogIndex, &d.BlockNumber,
		&d.AmountRaw, &d.USDValueMicros, &d.StarsCredited, &status, &d.Confirmations, &d.DetectedAt, &creditedAt, &created); err != nil {
		return domain.DonationDeposit{}, false, fmt.Errorf("record donation deposit: %w", err)
	}
	d.Status = domain.DonationDepositStatus(status)
	if creditedAt.Valid {
		d.CreditedAt = creditedAt.Time
	}
	return d, created, nil
}

func (s *DonationStore) UpdateDonationDepositConfirmations(ctx context.Context, chainKey, txHash string, logIndex int, confirmations int) (domain.DonationDeposit, error) {
	row := s.db.QueryRow(ctx, `
UPDATE donation_deposits SET
    confirmations = $4,
    status = CASE WHEN status = 'pending' AND $4 >= (
        SELECT confirmations_required FROM donation_chains WHERE chain_key = donation_deposits.chain_key
    ) THEN 'confirmed' ELSE status END
WHERE chain_key = $1 AND tx_hash = $2 AND log_index = $3 AND status = 'pending'
RETURNING `+donationDepositColumns, chainKey, txHash, logIndex, confirmations)
	out, err := scanDonationDeposit(row)
	if errors.Is(err, pgx.ErrNoRows) {
		return domain.DonationDeposit{}, nil // already past pending, or never existed; not an error
	}
	if err != nil {
		return domain.DonationDeposit{}, fmt.Errorf("update donation deposit confirmations: %w", err)
	}
	return out, nil
}

func (s *DonationStore) PendingDonationDeposits(ctx context.Context, chainKey string) ([]domain.DonationDeposit, error) {
	rows, err := s.db.Query(ctx, `SELECT `+donationDepositColumns+`
FROM donation_deposits WHERE chain_key = $1 AND status = 'pending' ORDER BY id`, chainKey)
	if err != nil {
		return nil, fmt.Errorf("list pending donation deposits: %w", err)
	}
	defer rows.Close()
	return scanDonationDeposits(rows)
}

func (s *DonationStore) ConfirmedUncreditedDeposits(ctx context.Context, chainKey string, limit int) ([]domain.DonationDeposit, error) {
	if limit <= 0 {
		limit = 50
	}
	rows, err := s.db.Query(ctx, `SELECT `+donationDepositColumns+`
FROM donation_deposits WHERE chain_key = $1 AND status = 'confirmed' ORDER BY id LIMIT $2`, chainKey, limit)
	if err != nil {
		return nil, fmt.Errorf("list confirmed donation deposits: %w", err)
	}
	defer rows.Close()
	return scanDonationDeposits(rows)
}

// CreditDonationDeposit is the only place a deposit's stars_credited is ever
// written, guarded by the WHERE status='confirmed' below: a deposit already
// credited (or somehow orphaned) matches zero rows, so a duplicate watcher
// pass or a retried call can never double-credit.
func (s *DonationStore) CreditDonationDeposit(ctx context.Context, depositID int64, usdValueMicros, stars int64, date int) (domain.DonationDeposit, bool, error) {
	if depositID <= 0 || stars <= 0 {
		return domain.DonationDeposit{}, false, domain.ErrDonationDepositInvalid
	}
	var credited bool
	err := withTx(ctx, s.db, "credit donation deposit", func(tx pgx.Tx) error {
		var userID int64
		var chainKey, tokenSymbol string
		err := tx.QueryRow(ctx, `
UPDATE donation_deposits SET status = 'credited', usd_value_micros = $2, stars_credited = $3, credited_at = now()
WHERE id = $1 AND status = 'confirmed'
RETURNING user_id, chain_key, token_symbol`, depositID, usdValueMicros, stars).Scan(&userID, &chainKey, &tokenSymbol)
		if errors.Is(err, pgx.ErrNoRows) {
			return nil // not confirmed (already credited, or never got there) -- not an error, just nothing to do.
		}
		if err != nil {
			return fmt.Errorf("mark donation deposit credited: %w", err)
		}
		symbol := tokenSymbol
		if symbol == "" {
			if chain, found, err := (&DonationStore{db: tx}).DonationChain(ctx, chainKey); err == nil && found {
				symbol = chain.NativeSymbol
			}
		}
		if _, err := tx.Exec(ctx, `
INSERT INTO stars_balances (user_id, balance, updated_at) VALUES ($1, $2, now())
ON CONFLICT (user_id) DO UPDATE SET balance = stars_balances.balance + EXCLUDED.balance, updated_at = now()`,
			userID, stars); err != nil {
			return fmt.Errorf("credit stars balance for donation: %w", err)
		}
		if err := insertStarsTxn(ctx, tx, userID, stars, domain.StarsReasonDonationDeposit, domain.Peer{}, date,
			"Crypto donation", fmt.Sprintf("%s deposit on %s", symbol, chainKey)); err != nil {
			return err
		}
		credited = true
		return nil
	})
	if err != nil {
		return domain.DonationDeposit{}, false, err
	}
	if !credited {
		return domain.DonationDeposit{}, false, nil
	}
	return domain.DonationDeposit{ID: depositID, Status: domain.DonationDepositCredited, USDValueMicros: usdValueMicros, StarsCredited: stars}, true, nil
}

func (s *DonationStore) UserDonationDeposits(ctx context.Context, userID int64, limit int) ([]domain.DonationDeposit, error) {
	if limit <= 0 || limit > 200 {
		limit = 50
	}
	rows, err := s.db.Query(ctx, `SELECT `+donationDepositColumns+`
FROM donation_deposits WHERE user_id = $1 ORDER BY detected_at DESC LIMIT $2`, userID, limit)
	if err != nil {
		return nil, fmt.Errorf("list user donation deposits: %w", err)
	}
	defer rows.Close()
	return scanDonationDeposits(rows)
}