diff options
| -rw-r--r-- | generalresearch/managers/thl/cashout_method.py | 12 | ||||
| -rw-r--r-- | generalresearch/managers/thl/ledger_manager/ledger.py | 31 | ||||
| -rw-r--r-- | generalresearch/managers/thl/ledger_manager/thl_ledger.py | 416 | ||||
| -rw-r--r-- | generalresearch/managers/thl/paypal.py | 403 | ||||
| -rw-r--r-- | generalresearch/managers/thl/session.py | 6 | ||||
| -rw-r--r-- | generalresearch/managers/thl/wallet/__init__.py | 17 | ||||
| -rw-r--r-- | generalresearch/managers/thl/wallet/approve.py | 16 | ||||
| -rw-r--r-- | generalresearch/models/thl/ledger.py | 7 | ||||
| -rw-r--r-- | generalresearch/models/thl/product.py | 22 | ||||
| -rw-r--r-- | generalresearch/models/thl/session.py | 37 | ||||
| -rw-r--r-- | generalresearch/models/thl/wallet/user_wallet.py | 45 | ||||
| -rw-r--r-- | tests/models/thl/test_product_userwalletconfig.py | 6 |
12 files changed, 882 insertions, 136 deletions
diff --git a/generalresearch/managers/thl/cashout_method.py b/generalresearch/managers/thl/cashout_method.py index 6e32a45..7d2366d 100644 --- a/generalresearch/managers/thl/cashout_method.py +++ b/generalresearch/managers/thl/cashout_method.py @@ -248,9 +248,6 @@ class CashoutMethodManager(PostgresManager): product = user.product supported_payout_types = copy(product.user_wallet_config.supported_payout_types) - if product.user_wallet_config.amt: - supported_payout_types.add(PayoutType.AMT) - user_scoped_payout_types = [PayoutType.PAYPAL, PayoutType.CASH_IN_MAIL] params = { "user_scoped_payout_types": [x.value for x in user_scoped_payout_types], @@ -273,15 +270,6 @@ class CashoutMethodManager(PostgresManager): raise ValueError(f"Unexpectedly large number of cashout_methods: {user=}") cms = [self.format_from_db(x, user=user) for x in res] - - # Only allow AMT if the BP is marked as AMT (already should have been - # filtered in query) - cms = [ - x - for x in cms - if (x.type == PayoutType.AMT and product.user_wallet_config.amt) - or (x.type != PayoutType.AMT) - ] return cms @staticmethod diff --git a/generalresearch/managers/thl/ledger_manager/ledger.py b/generalresearch/managers/thl/ledger_manager/ledger.py index bbf7bfa..3a02cdf 100644 --- a/generalresearch/managers/thl/ledger_manager/ledger.py +++ b/generalresearch/managers/thl/ledger_manager/ledger.py @@ -37,7 +37,6 @@ from generalresearch.models.thl.ledger import ( ) if TYPE_CHECKING: - from generalresearch.models.thl.ledger import UserLedgerTransactionType from generalresearch.pg_helper import PostgresConfig from generalresearch.redis_helper import RedisConfig @@ -121,9 +120,9 @@ class LedgerManagerBasePostgres(PostgresManager, RedisManager): filters.append("key = %(metadata_key)s") params["metadata_key"] = metadata_key if metadata_value is not None: - assert ( - metadata_key is not None - ), "cannot filter by metadata_value without metadata_key" + assert metadata_key is not None, ( + "cannot filter by metadata_value without metadata_key" + ) filters.append("value = %(metadata_value)s") params["metadata_value"] = metadata_value @@ -132,7 +131,6 @@ class LedgerManagerBasePostgres(PostgresManager, RedisManager): class LedgerTransactionManager(LedgerManagerBasePostgres): - def create_tx( self, entries: list[LedgerEntry], @@ -148,9 +146,9 @@ class LedgerTransactionManager(LedgerManagerBasePostgres): lastrowid) """ - assert ( - Permission.CREATE in self.permissions - ), "LedgerTransactionManager has insufficient Permissions" + assert Permission.CREATE in self.permissions, ( + "LedgerTransactionManager has insufficient Permissions" + ) if metadata is None: metadata = {} @@ -357,10 +355,18 @@ class LedgerTransactionManager(LedgerManagerBasePostgres): raise ValueError(f"Too many txs with this tag: {tag}") return {x["id"] for x in res} - def get_tx_by_tag(self, tag: str) -> list[LedgerTransaction]: + def get_txs_by_tag(self, tag: str) -> list[LedgerTransaction]: tx_ids = self.get_tx_ids_by_tag(tag=tag) return self.get_tx_by_ids(transaction_ids=tx_ids) + def get_tx_by_tag_if_exists(self, tag: str) -> LedgerTransaction | None: + tx_ids = self.get_tx_ids_by_tag(tag=tag) + if not tx_ids: + return None + if len(tx_ids) != 1: + raise ValueError(f"Two transactions found for tag: {tag}!") + return self.get_tx_by_id(transaction_id=next(iter(tx_ids))) + def get_tx_ids_by_tags(self, tags: list[str]) -> set[PositiveInt]: res = self.pg_config.execute_sql_query( query=""" @@ -799,7 +805,6 @@ class LedgerMetadataManager(LedgerManagerBasePostgres): class LedgerEntryManager(LedgerManagerBasePostgres): - def get_tx_entries_by_tx(self, transaction: LedgerTransaction) -> list[LedgerEntry]: return self.get_tx_entries_by_txs(transactions=[transaction]) @@ -831,9 +836,9 @@ class LedgerAccountManager(LedgerManagerBasePostgres): """ def create_account(self, account: LedgerAccount) -> LedgerAccount: - assert ( - Permission.CREATE in self.permissions - ), "LedgerManager does not have sufficient permissions" + assert Permission.CREATE in self.permissions, ( + "LedgerManager does not have sufficient permissions" + ) d = account.model_dump(mode="json") diff --git a/generalresearch/managers/thl/ledger_manager/thl_ledger.py b/generalresearch/managers/thl/ledger_manager/thl_ledger.py index e5a308f..f1a76c4 100644 --- a/generalresearch/managers/thl/ledger_manager/thl_ledger.py +++ b/generalresearch/managers/thl/ledger_manager/thl_ledger.py @@ -1,6 +1,7 @@ from __future__ import annotations import logging +from collections import defaultdict from collections.abc import Callable, Collection from datetime import UTC, datetime, timedelta from decimal import Decimal @@ -9,6 +10,7 @@ from uuid import UUID import numpy as np import pandas as pd +from generalresearch.models.thl.wallet.definitions import PayoutType from pydantic import AwareDatetime, PositiveInt from generalresearch.config import ( @@ -51,20 +53,21 @@ from generalresearch.models.thl.ledger import ( from generalresearch.models.thl.ledger import ( TransactionMetadataColumns as tmc, ) +from generalresearch.models.thl.payout import UserPayoutEvent +from generalresearch.models.thl.payout_format import format_payout_format from generalresearch.models.thl.product import Product -from generalresearch.models.thl.session import Status -from generalresearch.models.thl.wallet.definitions import PayoutType +from generalresearch.models.thl.session import Session, Status, Wall +from generalresearch.models.thl.user import User +from generalresearch.models.thl.wallet.user_wallet import ( + UserDisplayedWalletBalance, + UserLedgerWallet, + UserLedgerWallets, +) if TYPE_CHECKING: - - from generalresearch.models.thl.contest.contest import Contest, ContestWinner - from generalresearch.models.thl.contest.raffle import ( - ContestEntry, - ) - from generalresearch.models.thl.ledger import LedgerTransaction - from generalresearch.models.thl.payout import UserPayoutEvent - from generalresearch.models.thl.session import Session, Wall - from generalresearch.models.thl.user import User + from generalresearch.managers.thl.session import SessionManager + from generalresearch.models.thl.contest.contest import ContestWinner + logging.basicConfig() logger = logging.getLogger("LedgerManager") @@ -72,7 +75,6 @@ logger.setLevel(logging.INFO) class ThlLedgerManager(LedgerManager): - def get_account_or_create_user_wallet(self, user: User) -> LedgerAccount: """ TODO: In the future we could create a user wallet account with a @@ -81,12 +83,36 @@ class ThlLedgerManager(LedgerManager): """ assert user.user_id, "User must be saved" + assert self.currency is not None, "Must set currency" + account_type = AccountType.USER_WALLET account = LedgerAccount( display_name=f"User Wallet {user.uuid}", - qualified_name=f"{self.currency.value}:user_wallet:{user.uuid}", + qualified_name=f"{self.currency.value}:{account_type.value}:{user.uuid}", normal_balance=Direction.CREDIT, - account_type=AccountType.USER_WALLET, + account_type=account_type, + reference_type="user", + reference_uuid=user.uuid, + currency=self.currency, + ) + + return self.get_account_or_create(account=account) + + def get_account_or_create_user_attempt_credit(self, user: User) -> LedgerAccount: + """ + A wallet to hold a user's conditional credits which are + applied against future task earnings + """ + + assert user.user_id, "User must be saved" + assert self.currency is not None, "Must set currency" + account_type = AccountType.USER_ATTEMPT_CREDIT + + account = LedgerAccount( + display_name=f"User Attempt Credit {user.uuid}", + qualified_name=f"{self.currency.value}:{account_type.value}:{user.uuid}", + normal_balance=Direction.CREDIT, + account_type=account_type, reference_type="user", reference_uuid=user.uuid, currency=self.currency, @@ -253,9 +279,9 @@ class ThlLedgerManager(LedgerManager): if time_end is None: time_end = datetime.now(tz=UTC) - assert all( - isinstance(item, str) for item in account_uuids - ), "Must pass account_uuid as str" + assert all(isinstance(item, str) for item in account_uuids), ( + "Must pass account_uuid as str" + ) params = { "time_start": time_start, @@ -363,13 +389,15 @@ class ThlLedgerManager(LedgerManager): f = lambda: self.create_tx_bp_payment_(session=session, created=created) condition = generate_condition_bp_payment(session) - lock_key = f"{self.currency.value}:thl_session:{session.uuid}" + lock_key = f"{self.currency.value}:user_uuid:{session.user.uuid}" + flag_key = f"{self.currency.value}:bp_payment:{session.uuid}" return self.create_tx_protected( lock_key=lock_key, condition=condition, create_tx_func=f, skip_flag_check=force, + flag_key=flag_key, ) def create_tx_bp_payment_( @@ -452,7 +480,31 @@ class ThlLedgerManager(LedgerManager): amount=user_pay, ) ) - ext_description = f"BP & User Payment {session.uuid}" + if user.product.user_wallet_config.failed_attempt_credit_enabled: + settlement_amount = self.get_user_attempt_credit_to_settle( + user=user, user_pay=user_pay + ) + + if settlement_amount: + attempt_credit_account = ( + self.get_account_or_create_user_attempt_credit(user) + ) + + entries.extend( + [ + LedgerEntry( + direction=Direction.DEBIT, + account_uuid=attempt_credit_account.uuid, + amount=settlement_amount, + ), + LedgerEntry( + direction=Direction.CREDIT, + account_uuid=bp_wallet_account.uuid, + amount=settlement_amount, + ), + ] + ) + ext_description = f"BP & User Payment {session.uuid}" else: entries.append( @@ -995,8 +1047,8 @@ class ThlLedgerManager(LedgerManager): user: User, payout_event: UserPayoutEvent, created: datetime | None = None, - skip_flag_check: bool | None = False, - skip_wallet_balance_check: bool | None = False, + skip_flag_check: bool = False, + skip_wallet_balance_check: bool = False, ) -> LedgerTransaction: """ The funds move from the user's wallet into the BP's "pending" @@ -1008,35 +1060,19 @@ class ThlLedgerManager(LedgerManager): requesting from their USD wallet. No other currencies are supported now. """ - assert ( - user.product.user_wallet_enabled - ), "Can only call this on an wallet enabled BPs" + assert user.product.user_wallet_enabled, ( + "Can only call this on an wallet enabled BPs" + ) amount = USDCent(payout_event.amount) amt_str = f"${int(amount) / 100:,.2f}" descriptions = { - PayoutType.AMT_HIT: f"User Payout AMT Assignment Request {amt_str}", - PayoutType.AMT_BONUS: f"User Payout AMT Bonus Request {amt_str}", PayoutType.PAYPAL: f"User Payout Paypal Request {amt_str}", PayoutType.CASH_IN_MAIL: f"User Payout Cash Request {amt_str}", PayoutType.TANGO: f"User Payout Tango Request {amt_str}", } description = descriptions[payout_event.payout_type] - if payout_event.payout_type in { - PayoutType.AMT_HIT, - PayoutType.AMT_BONUS, - }: - """ - This is for AMT accounts only (currently JB). This is the - payment of a either 1) 1c or 5c (typically) assignment or 2) a - bonus for task complete to the user. The 20% commission will - be taken from the BP's wallet once the tx is completed. - """ - assert ( - user.product.user_wallet_amt - ), "Can only call this on an AMT-enabled BPs" - f = lambda: self.create_tx_user_payout_request_( user=user, payout_event=payout_event, @@ -1045,10 +1081,6 @@ class ThlLedgerManager(LedgerManager): ) min_balance: int | None = int(amount) - if payout_event.payout_type == PayoutType.AMT_HIT: - # We allow the user's balance to reach up to -$1.00. - min_balance = -100 + amount - if skip_wallet_balance_check: min_balance = None @@ -1082,9 +1114,9 @@ class ThlLedgerManager(LedgerManager): are taken from the BP's pending wallet, the commission will be recorded, and the cash debited. """ - assert ( - user.product.user_wallet_enabled - ), "Can only call this on an wallet enabled BPs" + assert user.product.user_wallet_enabled, ( + "Can only call this on an wallet enabled BPs" + ) # Before we even do anything, we should check that a ledger tx exists for the request request_tag = f"{self.currency.value}:user_payout:{payout_event.uuid}:request" @@ -1110,9 +1142,9 @@ class ThlLedgerManager(LedgerManager): PayoutType.AMT_HIT, PayoutType.AMT_BONUS, }: - assert ( - user.product.user_wallet_amt - ), "Can only call this on an AMT-enabled BP" + assert user.product.user_wallet_amt, ( + "Can only call this on an AMT-enabled BP" + ) bp_expense_account = self.get_account_or_create_bp_expense( product=user.product, expense_name="amt" ) @@ -1174,9 +1206,9 @@ class ThlLedgerManager(LedgerManager): created: datetime | None = None, skip_flag_check: bool | None = False, ) -> LedgerTransaction: - assert ( - user.product.user_wallet_enabled - ), "Can only call this on an wallet enabled BPs" + assert user.product.user_wallet_enabled, ( + "Can only call this on an wallet enabled BPs" + ) # Before we even do anything, we should check that a ledger tx exists for the request request_tag = f"{self.currency.value}:user_payout:{payout_event.uuid}:request" @@ -1401,9 +1433,9 @@ class ThlLedgerManager(LedgerManager): :param source_account: Is this paid from the bp's wallet? or from us? """ - assert ( - user.product.user_wallet_enabled - ), "Can only call this on an wallet enabled BPs" + assert user.product.user_wallet_enabled, ( + "Can only call this on an wallet enabled BPs" + ) assert user.product, "user.prefetch_product()" # This tag should uniquely id this tx. @@ -1472,6 +1504,120 @@ class ThlLedgerManager(LedgerManager): created=created, ) + def create_tx_attempt_credit( + self, + session: Session, + created: datetime | None = None, + skip_flag_check: bool = False, + ) -> LedgerTransaction: + """Record conditional credit for an eligible session attempt. + + Eligibility is determined by the caller. This method verifies that the + session failed and records the credit exactly once per session. + """ + user = session.user + assert user is not None, "Session must have a user" + product = user.product + assert product is not None, "user.prefetch_product()" + assert product.user_wallet_enabled, "Product does not have user_wallet enabled" + config = product.user_wallet_config + assert config.failed_attempt_credit_enabled, ( + "Product does not have failed_attempt_credit enabled" + ) + amount = USDCent(round(config.failed_attempt_credit * 100)) + + assert session.is_attempt_credit_eligible(), ( + "Session is not eligible for attempt credit" + ) + + tag = f"{self.currency.value}:{TransactionType.USER_ATTEMPT_CREDIT.value}:{session.uuid}" + condition = generate_condition_tag_exists(tag) + + def create() -> LedgerTransaction: + bp_account = self.get_account_or_create_bp_wallet(product) + attempt_credit_account = self.get_account_or_create_user_attempt_credit( + user + ) + metadata = { + tmc.USER.value: user.uuid, + tmc.SESSION.value: session.uuid, + tmc.TX_TYPE.value: TransactionType.USER_ATTEMPT_CREDIT.value, + } + entries = [ + LedgerEntry( + direction=Direction.DEBIT, + account_uuid=bp_account.uuid, + amount=amount, + ), + LedgerEntry( + direction=Direction.CREDIT, + account_uuid=attempt_credit_account.uuid, + amount=amount, + ), + ] + return self.create_tx( + entries=entries, + metadata=metadata, + tag=tag, + ext_description=f"Attempt Credit {session.uuid}", + created=created, + ) + + return self.create_tx_protected( + lock_key=f"{self.currency.value}:user_uuid:{user.uuid}", + flag_key=tag, + condition=condition, + create_tx_func=create, + skip_flag_check=skip_flag_check, + ) + + def claim_latest_attempt_credit( + self, + user: User, + session_manager: SessionManager, + skip_flag_check: bool = False, + ) -> LedgerTransaction: + """Claim attempt credit for a product user's most recent session. + This must be for an abandoned session, as when a session if finished + and is eligible for an attempt credit, then the credit is automatically + given.""" + assert user.user_id is not None + session = session_manager.get_latest_for_user(user_id=user.user_id) + if session is None: + raise ValueError("User has no session to claim attempt credit for") + if session.status is not None: + raise ValueError("User's latest session is already finalized") + + assert user.product + session.user.product = user.product + return self.create_tx_attempt_credit( + session=session, + skip_flag_check=skip_flag_check, + ) + + def get_session_attempt_credit( + self, + session_uuid: UUIDStr, + ) -> int | None: + tag = ( + f"{self.currency.value}:" + f"{TransactionType.USER_ATTEMPT_CREDIT.value}:" + f"{session_uuid}" + ) + tx = self.get_tx_by_tag_if_exists(tag) + if not tx: + return None + + credit_entries = [ + entry for entry in tx.entries if entry.direction == Direction.CREDIT + ] + if len(credit_entries) != 1: + raise ValueError( + f"Expected one credit entry for attempt-credit transaction {tx.id}" + ) + + return credit_entries[0].amount + def create_tx_user_enter_contest( self, contest_uuid: UUIDStr, @@ -1482,13 +1628,13 @@ class ThlLedgerManager(LedgerManager): User is requesting to enter a Raffle Contest. We'll DEBIT funds from their wallet and CREDIT the contest wallet. """ - assert ( - contest_entry.entry_type == ContestEntryType.CASH - ), "Can only call this for CASH Contests" + assert contest_entry.entry_type == ContestEntryType.CASH, ( + "Can only call this for CASH Contests" + ) user = contest_entry.user - assert ( - user.product.user_wallet_enabled - ), "Can only call this on an wallet enabled BPs" + assert user.product.user_wallet_enabled, ( + "Can only call this on an wallet enabled BPs" + ) assert user.product, "user.prefetch_product()" amount = contest_entry.amount entry_uuid = contest_entry.uuid @@ -1572,9 +1718,9 @@ class ThlLedgerManager(LedgerManager): Any remaining money goes back into the BP's wallet ? todo """ if contest.contest_type in {ContestType.RAFFLE, ContestType.MILESTONE}: - assert ( - contest.entry_type == ContestEntryType.CASH - ), "Can only call this for CASH Contests" + assert contest.entry_type == ContestEntryType.CASH, ( + "Can only call this for CASH Contests" + ) contest_account = self.get_account_or_create_contest_wallet_by_uuid( contest_uuid=contest.uuid @@ -1829,9 +1975,9 @@ class ThlLedgerManager(LedgerManager): :returns wallet balance in integer cents """ user.prefetch_product(self.pg_config) - assert ( - user.product.user_wallet_config.enabled - ), "Can't get wallet balance on non-managed account." + assert user.product.user_wallet_config.enabled, ( + "Can't get wallet balance on non-managed account." + ) now = datetime.now(tz=UTC) wallet = self.get_account_or_create_user_wallet(user) @@ -1886,7 +2032,9 @@ class ThlLedgerManager(LedgerManager): wall["user_payout"] = wall["user_payout"].astype(float) wall["user_payout_int"] = wall["user_payout"] * 100 wall["days_since_complete"] = (now - wall["finished"]).dt.days - wall["pct_rdm"] = wall["days_since_complete"].apply(self.get_redeemable_pct) + wall["pct_rdm"] = wall["days_since_complete"].apply( + self._get_redeemable_pct + ) wall.loc[wall["pct_rdm"] > 0.95, "pct_rdm"] = 1 wall["redeemable"] = wall["pct_rdm"] * wall["user_payout_int"] # Calculate money needed to save in reserve to cover the difference @@ -1903,7 +2051,7 @@ class ThlLedgerManager(LedgerManager): assert redeemable_balance <= user_wallet_balance return redeemable_balance - def get_redeemable_pct( + def _get_redeemable_pct( self, days_since_complete: float, user_trust: float = 0.0 ) -> float: """ @@ -1928,6 +2076,134 @@ class ThlLedgerManager(LedgerManager): # plt.plot([d.days for d in x], [self.get_redeemable_amount(d) for d in x]) return pct_rdm + def get_user_attempt_credit_balance(self, user: User) -> int: + """Return the user's outstanding conditional attempt credit in cents.""" + assert user.user_id, "User must be saved" + account = self.get_account_or_create_user_attempt_credit(user) + return self.get_account_balance(account) + + def get_user_wallets(self, user: User) -> UserLedgerWallets: + """Return every ledger wallet owned by a user, across currencies.""" + assert user.user_id, "User must be saved" + user.prefetch_product(self.pg_config) + payout_format = user.product.payout_config.payout_format + assert payout_format is not None, "Product must have a payout format" + user_account_types = { + AccountType.USER_WALLET.value, + AccountType.USER_ATTEMPT_CREDIT.value, + } + + rows = self.pg_config.execute_sql_query( + query=""" + SELECT + uuid, display_name, qualified_name, account_type, + normal_balance, reference_type, reference_uuid, currency + FROM ledger_account + WHERE reference_type = 'user' + AND reference_uuid = %(reference_uuid)s + AND account_type = ANY(%(account_types)s) + ORDER BY currency, account_type, qualified_name; + """, + params={ + "reference_uuid": user.uuid, + "account_types": list(user_account_types), + }, + ) + accounts = [LedgerAccount.model_validate(row) for row in rows] + + wallets = [] + displayed_amounts: dict[str, int] = defaultdict(int) + for account in accounts: + if ( + user.product_id == JAMES_BILLINGS_BPID + and account.account_type == AccountType.USER_WALLET + ): + balance = self.get_account_balance_timerange( + account=account, + time_start=JAMES_BILLINGS_TX_CUTOFF, + ) + else: + balance = self.get_account_balance(account) + + if account.account_type == AccountType.USER_ATTEMPT_CREDIT: + redeemable_balance = 0 + elif account.currency == self.currency.value: + redeemable_balance = self.get_user_redeemable_wallet_balance( + user=user, + user_wallet_balance=balance, + ) + else: + # We don't have redeemable logic for other currencies + redeemable_balance = max(balance, 0) + + # A Product has only one payout_format. It is not clear which + # currency it is to be applied to. If we have a non USD currency, + # we'd need multiple payout formats. + account_payout_format = ( + payout_format if account.currency == self.currency.value else None + ) + + wallet = UserLedgerWallet( + account_uuid=account.uuid, + account_type=account.account_type, + currency=account.currency, + display_name=account.display_name, + amount=balance, + redeemable_amount=redeemable_balance, + payout_format=account_payout_format, + amount_string=( + format_payout_format(account_payout_format, balance) + if account_payout_format is not None + else None + ), + redeemable_amount_string=( + format_payout_format(account_payout_format, redeemable_balance) + if account_payout_format is not None + else None + ), + ) + wallets.append(wallet) + + if ( + account.account_type == AccountType.USER_ATTEMPT_CREDIT + or user.product.user_wallet_config.balance_type == "wallet_balance" + ): + displayed_amount = wallet.amount + else: + displayed_amount = wallet.redeemable_amount + displayed_amounts[account.currency] += displayed_amount + + displayed_balances = [ + UserDisplayedWalletBalance( + currency=currency, + amount=amount, + amount_string=( + format_payout_format(payout_format, amount) + if currency == self.currency.value + else None + ), + ) + for currency, amount in sorted(displayed_amounts.items()) + ] + + return UserLedgerWallets( + wallets=wallets, + displayed_balances=displayed_balances, + ) + + def get_user_attempt_credit_to_settle( + self, + user: User, + user_pay: int, + ) -> int: + """Return how much pending credit a task payment of amount `user_pay` should consume.""" + assert user_pay >= 0 + + credit_balance = self.get_user_attempt_credit_balance(user) + assert credit_balance >= 0, "Attempt-credit balance cannot be negative" + + return min(credit_balance, user_pay) + def get_user_txs( self, user: User, diff --git a/generalresearch/managers/thl/paypal.py b/generalresearch/managers/thl/paypal.py new file mode 100644 index 0000000..5ef0966 --- /dev/null +++ b/generalresearch/managers/thl/paypal.py @@ -0,0 +1,403 @@ +from __future__ import annotations + +from datetime import datetime, timedelta, timezone +from decimal import Decimal +from typing import Any, Mapping + +import requests + +from generalresearch.currency import USDCent +from generalresearch.managers.thl.ledger_manager.thl_ledger import ( + ThlLedgerManager, +) +from generalresearch.managers.thl.payout import UserPayoutEventManager +from generalresearch.managers.thl.user_manager.user_manager import UserManager +from generalresearch.models.thl.definitions import PayoutStatus +from generalresearch.models.thl.payout import UserPayoutEvent +from generalresearch.models.thl.user import User +from generalresearch.models.thl.wallet import PayoutType +from generalresearch.models.thl.wallet.cashout_method import ( + CashoutMethod, + PaypalCashoutMethodData, +) + +PAYPAL_SANDBOX_URL = "https://api-m.sandbox.paypal.com" +PAYPAL_PROD_URL = "https://api-m.paypal.com" + + +class PayPalError(RuntimeError): + """Raised when PayPal returns an unsuccessful API response.""" + + +class PayPalPayoutManager: + """Small client for PayPal's Payouts REST API. + + Credentials are supplied by the caller so they can come from the + application's secret store. ``sender_batch_id`` and ``sender_item_id`` + should be stable application identifiers; PayPal uses them to prevent + duplicate payouts. + """ + + def __init__( + self, + *, + client_id: str, + client_secret: str, + base_url: str = PAYPAL_SANDBOX_URL, + timeout: float = 30.0, + session: requests.Session | None = None, + ) -> None: + if not client_id or not client_secret: + raise ValueError("PayPal client_id and client_secret are required") + + self.client_id = client_id + self.client_secret = client_secret + self.base_url = base_url.rstrip("/") + self.timeout = timeout + self.session = session or requests.Session() + self._access_token: str | None = None + self._access_token_expires_at: datetime | None = None + + def _get_access_token(self) -> str: + now = datetime.now(tz=timezone.utc) + if ( + self._access_token + and self._access_token_expires_at + and now < self._access_token_expires_at + ): + return self._access_token + + response = self.session.post( + f"{self.base_url}/v1/oauth2/token", + auth=(self.client_id, self.client_secret), + data={"grant_type": "client_credentials"}, + headers={"Accept": "application/json"}, + timeout=self.timeout, + ) + data = self._response_json(response) + token = data.get("access_token") + if not token: + raise PayPalError("PayPal token response did not include access_token") + + # Refresh one minute early to avoid using a token while it expires. + expires_in = max(int(data.get("expires_in", 300)) - 60, 0) + self._access_token = token + self._access_token_expires_at = now + timedelta(seconds=expires_in) + return token + + def _request( + self, + method: str, + path: str, + *, + json: dict[str, Any] | None = None, + params: dict[str, Any] | None = None, + ) -> dict[str, Any]: + response = self.session.request( + method, + f"{self.base_url}{path}", + headers={ + "Accept": "application/json", + "Content-Type": "application/json", + "Authorization": f"Bearer {self._get_access_token()}", + }, + json=json, + params=params, + timeout=self.timeout, + ) + return self._response_json(response) + + @staticmethod + def _response_json(response: requests.Response) -> dict[str, Any]: + try: + data = response.json() + except ValueError: + data = {"message": response.text} + + if not response.ok: + debug_id = response.headers.get("PayPal-Debug-Id") + raise PayPalError( + f"PayPal API error {response.status_code}" + + (f" PayPal-Debug-Id={debug_id}" if debug_id else "") + + f": {data}" + ) + if not isinstance(data, dict): + raise PayPalError(f"Unexpected PayPal response: {data!r}") + return data + + def send_payment( + self, + *, + recipient_email: str, + amount_cents: USDCent, + sender_batch_id: str, + sender_item_id: str | None = None, + note: str | None = None, + email_subject: str | None = None, + ) -> dict[str, Any]: + """Create a one-recipient USD payout and return PayPal's response.""" + assert isinstance(amount_cents, USDCent) + if amount_cents <= 0: + raise ValueError("amount_cents must be positive") + if not recipient_email: + raise ValueError("recipient_email is required") + if not sender_batch_id: + raise ValueError("sender_batch_id is required") + + amount = (Decimal(amount_cents) / 100).quantize(Decimal("0.01")) + batch_header: dict[str, Any] = { + "sender_batch_id": sender_batch_id, + "recipient_type": "EMAIL", + } + if email_subject: + batch_header["email_subject"] = email_subject + + item: dict[str, Any] = { + "recipient_type": "EMAIL", + "receiver": recipient_email, + "amount": {"value": f"{amount:.2f}", "currency": "USD"}, + "sender_item_id": sender_item_id or sender_batch_id, + } + if note: + item["note"] = note + + return self._request( + "POST", + "/v1/payments/payouts", + json={"sender_batch_header": batch_header, "items": [item]}, + ) + + def get_payout(self, payout_batch_id: str) -> dict[str, Any]: + """Return the latest batch and item status for a payout.""" + if not payout_batch_id: + raise ValueError("payout_batch_id is required") + return self._request("GET", f"/v1/payments/payouts/{payout_batch_id}") + + def verify_webhook_signature( + self, + *, + headers: Mapping[str, str], + webhook_event: dict[str, Any], + webhook_id: str, + ) -> bool: + """Ask PayPal to verify a webhook notification's signature.""" + normalized_headers = {key.lower(): value for key, value in headers.items()} + + def required_header(name: str) -> str: + value = normalized_headers.get(name.lower()) + if not value: + raise ValueError(f"Missing PayPal webhook header: {name}") + return value + + result = self._request( + "POST", + "/v1/notifications/verify-webhook-signature", + json={ + "auth_algo": required_header("PAYPAL-AUTH-ALGO"), + "cert_url": required_header("PAYPAL-CERT-URL"), + "transmission_id": required_header("PAYPAL-TRANSMISSION-ID"), + "transmission_sig": required_header("PAYPAL-TRANSMISSION-SIG"), + "transmission_time": required_header("PAYPAL-TRANSMISSION-TIME"), + "webhook_id": webhook_id, + "webhook_event": webhook_event, + }, + ) + return result.get("verification_status") == "SUCCESS" + + +def create_paypal_payout( + *, + user: User, + cashout_method: CashoutMethod, + amount_cents: USDCent, + user_payout_event_manager: UserPayoutEventManager, + ledger_manager: ThlLedgerManager, + paypal: PayPalPayoutManager, + note: str | None = None, + email_subject: str | None = None, +) -> UserPayoutEvent: + """Create, reserve funds for, and submit a one-item PayPal payout. + + The payout-event UUID is used as both sender IDs, making a retry at + PayPal idempotent for 30 days. The PayPal-generated batch ID is persisted + in ``ext_ref_id`` and the initial API response is kept in ``order_data``. + + If submission raises, the payout event and its ledger reservation remain + in place. This is intentional: a timeout or 5xx response is ambiguous and + rolling back could allow the same money to be paid twice. The exception + includes the payout-event UUID for reconciliation or a same-ID retry. + """ + if cashout_method.type != PayoutType.PAYPAL: + raise ValueError("cashout_method must be a PayPal cashout method") + if not isinstance(cashout_method.data, PaypalCashoutMethodData): + raise ValueError("cashout_method does not contain PayPal data") + if ( + user.user_id is None + or cashout_method.user is None + or cashout_method.user.user_id != user.user_id + ): + raise ValueError("cashout_method does not belong to user") + + cashout_method.validate_requested_amount(amount_cents) + user_account = ledger_manager.get_account_or_create_user_wallet(user=user) + payout_event = user_payout_event_manager.create( + debit_account_uuid=user_account.uuid, + cashout_method_uuid=cashout_method.id, + payout_type=PayoutType.PAYPAL, + amount=amount_cents, + status=PayoutStatus.PENDING, + account_reference_type="user", + account_reference_uuid=user.uuid, + description=cashout_method.name, + request_data={ + "interface": "api", + "recipient_email": str(cashout_method.data.email), + }, + ) + ledger_manager.create_tx_user_payout_request( + user=user, + payout_event=payout_event, + ) + + try: + paypal_response = paypal.send_payment( + recipient_email=str(cashout_method.data.email), + amount_cents=amount_cents, + sender_batch_id=payout_event.uuid, + sender_item_id=payout_event.uuid, + note=note, + email_subject=email_subject, + ) + payout_batch_id = paypal_response["batch_header"]["payout_batch_id"] + except Exception as exc: + raise PayPalError( + f"PayPal submission failed for payout event {payout_event.uuid}" + ) from exc + + user_payout_event_manager.update( + payout_event=payout_event, + status=PayoutStatus.PENDING, + ext_ref_id=payout_batch_id, + order_data=paypal_response, + ) + return payout_event + + +def handle_paypal_payout_webhook( + *, + headers: Mapping[str, str], + webhook_event: dict[str, Any], + webhook_id: str, + paypal: PayPalPayoutManager, + user_payout_event_manager: UserPayoutEventManager, + user_manager: UserManager, + ledger_manager: ThlLedgerManager, +) -> UserPayoutEvent | None: + """Verify and apply a PayPal Payouts webhook idempotently.""" + if not paypal.verify_webhook_signature( + headers=headers, + webhook_event=webhook_event, + webhook_id=webhook_id, + ): + raise PayPalError("PayPal webhook signature verification failed") + + event_type = webhook_event.get("event_type", "") + if not event_type.startswith(("PAYMENT.PAYOUTSBATCH.", "PAYMENT.PAYOUTS-ITEM.")): + return None + + resource = webhook_event.get("resource") + if not isinstance(resource, dict): + raise PayPalError("PayPal payout webhook has no resource") + + sender_batch_header = resource.get("sender_batch_header") or {} + payout_item = resource.get("payout_item") or {} + payout_event_uuid = ( + resource.get("sender_batch_id") + or sender_batch_header.get("sender_batch_id") + or payout_item.get("sender_item_id") + ) + if not payout_event_uuid: + raise PayPalError("PayPal payout webhook has no sender payout ID") + + payout_event = user_payout_event_manager.get_by_uuid(payout_event_uuid) + if payout_event.payout_type != PayoutType.PAYPAL: + raise PayPalError(f"Payout event {payout_event.uuid} is not a PayPal payout") + if payout_event.status == PayoutStatus.COMPLETE: + return payout_event + + payout_batch_id = resource.get("payout_batch_id") or payout_event.ext_ref_id + if not payout_batch_id: + raise PayPalError("PayPal payout webhook has no payout_batch_id") + if payout_event.ext_ref_id and payout_event.ext_ref_id != payout_batch_id: + raise PayPalError("PayPal payout_batch_id does not match the payout event") + + details = paypal.get_payout(payout_batch_id) + items = details.get("items") or [] + if len(items) != 1: + raise PayPalError( + f"Expected one PayPal payout item for {payout_event.uuid}, got {len(items)}" + ) + item = items[0] + item_sender_id = (item.get("payout_item") or {}).get("sender_item_id") + if item_sender_id != payout_event.uuid: + raise PayPalError("PayPal sender_item_id does not match the payout event") + + amount = (item.get("payout_item") or {}).get("amount") or {} + if amount.get("currency") != "USD": + raise PayPalError("Only USD PayPal payouts are supported") + amount_cents = int(Decimal(amount["value"]) * 100) + if amount_cents != payout_event.amount: + raise PayPalError("PayPal payout amount does not match the payout event") + + # get_payout_detail() expects transaction_id at the top level. + details["transaction_id"] = item.get("transaction_id") + transaction_status = item.get("transaction_status") + if transaction_status == "SUCCESS": + fee = item.get("payout_item_fee") or {} + if fee.get("currency") != "USD" or "value" not in fee: + raise PayPalError("Successful PayPal payout has no USD fee") + + user = user_manager.get_user(user_uuid=payout_event.account_reference_uuid) + user.prefetch_product(pg_config=ledger_manager.pg_config) + complete_tag = ( + f"{ledger_manager.currency.value}:user_payout:" + f"{payout_event.uuid}:complete" + ) + complete_transactions = ledger_manager.get_tx_ids_by_tag(complete_tag) + if len(complete_transactions) > 1: + raise PayPalError(f"Multiple ledger transactions found for {complete_tag}") + if not complete_transactions: + ledger_manager.create_tx_user_payout_complete( + user=user, + payout_event=payout_event, + fee_amount=Decimal(fee["value"]), + ) + user_payout_event_manager.update( + payout_event=payout_event, + status=PayoutStatus.COMPLETE, + ext_ref_id=payout_batch_id, + order_data=details, + ) + elif transaction_status in { + "FAILED", + "BLOCKED", + "RETURNED", + "REFUNDED", + "REVERSED", + "CANCELED", + }: + user_payout_event_manager.update( + payout_event=payout_event, + status=PayoutStatus.FAILED, + ext_ref_id=payout_batch_id, + order_data=details, + ) + elif payout_event.status == PayoutStatus.PENDING: + user_payout_event_manager.update( + payout_event=payout_event, + status=PayoutStatus.APPROVED, + ext_ref_id=payout_batch_id, + order_data=details, + ) + + return payout_event diff --git a/generalresearch/managers/thl/session.py b/generalresearch/managers/thl/session.py index 5007f43..3f235fd 100644 --- a/generalresearch/managers/thl/session.py +++ b/generalresearch/managers/thl/session.py @@ -117,6 +117,12 @@ class SessionManager(PostgresManager): assert len(res) == 1 return self.session_from_mysql(res[0]) + def get_latest_for_user(self, user_id: int) -> Session | None: + """Return the most recently started session for a product user.""" + res, _ = self.filter_paginated(user_id=user_id, order_by="-started", size=1) + if res: + return res[0] + def get_from_id(self, session_id: int) -> Session: query = """ SELECT diff --git a/generalresearch/managers/thl/wallet/__init__.py b/generalresearch/managers/thl/wallet/__init__.py index f805872..7a8b4c9 100644 --- a/generalresearch/managers/thl/wallet/__init__.py +++ b/generalresearch/managers/thl/wallet/__init__.py @@ -2,7 +2,6 @@ from decimal import Decimal from typing import TYPE_CHECKING, Any from generalresearch.managers.thl.wallet.approve import ( - approve_amt_cashout, approve_paypal_order, ) from generalresearch.models.thl.definitions import PayoutStatus @@ -91,14 +90,6 @@ def manage_pending_cashout( payout_event=pe, payout_event_manager=payout_event_manager ) - elif pe.payout_type in {PayoutType.AMT_BONUS, PayoutType.AMT_HIT}: - approve_amt_cashout( - user=user, - payout_event=pe, - payout_event_manager=payout_event_manager, - ledger_manager=ledger_manager, - ) - elif pe.payout_type == PayoutType.CASH_IN_MAIL: assert order_data, "must pass order_data" payout_event_manager.update( @@ -116,14 +107,8 @@ def manage_pending_cashout( return pe elif new_status == PayoutStatus.COMPLETE: - # Used only for AMT/dummy cashouts that are actually paid out not - # by us. They are informing us that the cashout was successfully - # sent to the user - if pe.payout_type in {PayoutType.AMT_BONUS, PayoutType.AMT_HIT}: - # We already do this under approve_amt_cashout() - pass - elif pe.payout_type == PayoutType.PAYPAL: + if pe.payout_type == PayoutType.PAYPAL: # This is an issue here in that we actually don't know what the # fee is until it is sent and we read it back from paypal's csv # result. We have to just run this with a custom script, which diff --git a/generalresearch/managers/thl/wallet/approve.py b/generalresearch/managers/thl/wallet/approve.py index 7fedec1..7cae025 100644 --- a/generalresearch/managers/thl/wallet/approve.py +++ b/generalresearch/managers/thl/wallet/approve.py @@ -40,19 +40,3 @@ def approve_paypal_order( return payout_event -def approve_amt_cashout( - user: User, - payout_event: UserPayoutEvent, - ledger_manager: ThlLedgerManager, - payout_event_manager: PayoutEventManager, -) -> None: - """ - This is going to be paid out by the requester (the jb-lambdas) as an AMT bonus. - """ - assert payout_event.status in { - PayoutStatus.PENDING, - PayoutStatus.FAILED, - }, "attempting to manage payout that is not pending (or you can retry a failed order)" - - payout_event_manager.update(payout_event, status=PayoutStatus.APPROVED) - ledger_manager.create_tx_user_payout_complete(user, payout_event=payout_event) diff --git a/generalresearch/models/thl/ledger.py b/generalresearch/models/thl/ledger.py index 2290a83..613d16e 100644 --- a/generalresearch/models/thl/ledger.py +++ b/generalresearch/models/thl/ledger.py @@ -102,8 +102,10 @@ class AccountType(StrEnum, metaclass=ReprEnumMeta): BP_COMMISSION = "bp_commission" # BP wallets (owed balance) BP_WALLET = "bp_wallet" - # User's wallet + # User's wallet: real earnings potentially available for redemption USER_WALLET = "user_wallet" + # User's conditional credits applied against future task earnings + USER_ATTEMPT_CREDIT = "user_attempt_credit" # Cash account CASH = "cash" # Revenue (money coming in) @@ -168,6 +170,9 @@ class TransactionType(StrEnum): # might be called "expenses" in finance reports). USER_BONUS = "user_bonus" + # A user is paid conditional credits applied against future task earnings + USER_ATTEMPT_CREDIT = "user_attempt_credit" + # A transaction is made to plug accounting imbalances PLUG = "plug" diff --git a/generalresearch/models/thl/product.py b/generalresearch/models/thl/product.py index 3fd149d..7441318 100644 --- a/generalresearch/models/thl/product.py +++ b/generalresearch/models/thl/product.py @@ -412,7 +412,12 @@ class UserWalletConfig(BaseModel): default=False, description="If enabled, the users' wallets are managed." ) - # This field could go in supported_payout_types ---v + balance_type: Literal["wallet_balance", "redeemable_balance"] = Field( + default="wallet_balance", + description="The balance used as the user's displayed balance.", + ) + + # Deprecated amt: bool = Field(default=False, description="Uses Amazon Mechanical Turk") supported_payout_types: set[PayoutType] = Field( @@ -427,13 +432,23 @@ class UserWalletConfig(BaseModel): examples=[Decimal("10.00")], ) + failed_attempt_credit: Decimal | None = Field( + default=None, + gt=0, + description=( + "Conditional credit awarded for an eligible failed attempt. " + "None disables failed-attempt credits." + ), + examples=[Decimal("0.05"), None] + ) + @field_serializer("supported_payout_types", when_used="json") def serialize_supported_payout_types_in_order( self, supported_payout_types: set[PayoutType] ) -> list[PayoutType]: return sorted(supported_payout_types) - @field_validator("min_cashout", mode="after") + @field_validator("min_cashout", "failed_attempt_credit", mode="after") @classmethod def check_payout_decimal_places(cls, v: Decimal) -> Decimal: if v is not None: @@ -457,6 +472,9 @@ class UserWalletConfig(BaseModel): self.min_cashout = Decimal("0.01") return self + @property + def failed_attempt_credit_enabled(self) -> bool: + return self.failed_attempt_credit is not None class PayoutTransformationPercentArgs(BaseModel): pct: NonNegativeFloat = Field( diff --git a/generalresearch/models/thl/session.py b/generalresearch/models/thl/session.py index b3cb6af..a9fe442 100644 --- a/generalresearch/models/thl/session.py +++ b/generalresearch/models/thl/session.py @@ -434,9 +434,9 @@ class Wall(WallBase): # --- Properties --- - @computed_field + @computed_field() @property - def elapsed(self) -> timedelta: + def elapsed(self) -> timedelta | None: return self.finished - self.started if self.finished else None def to_json(self) -> str: @@ -819,12 +819,39 @@ class Session(BaseModel): self.model_config["validate_assignment"] = True self.__class__.model_validate(self) - def model_dump_mysql(self) -> dict[str, str | int | datetime | float | None]: + def is_attempt_credit_eligible(self) -> bool: + """Return whether this session qualifies for attempt credit. + The status of the BP's user_wallet_config.failed_attempt_credit_enabled does + not matter here. + """ + now = datetime.now(tz=UTC) + min_session_length = timedelta(minutes=1) + + if self.status is None: + return now - self.started >= min_session_length + + ineligible_status_codes = { + StatusCode1.SESSION_START_FAIL, + StatusCode1.SESSION_START_QUALITY_FAIL, + StatusCode1.SESSION_CONTINUE_QUALITY_FAIL, + StatusCode1.BUYER_QUALITY_FAIL, + StatusCode1.PS_BLOCKED, + StatusCode1.PS_QUALITY, + } + return ( + self.status == Status.FAIL + and self.status_code_1 not in ineligible_status_codes + and self.finished is not None + and self.elapsed >= min_session_length + ) + + def model_dump_mysql( + self, **kwargs + ) -> dict[str, str | int | datetime | float | None]: # Generate a dictionary representation of the model, with special # handling for datetimes, and nested models such as User & Bucket - - d = self.model_dump(mode="json") + d = self.model_dump(mode="json", **kwargs) d["started"] = self.started.replace(tzinfo=None) if self.finished: diff --git a/generalresearch/models/thl/wallet/user_wallet.py b/generalresearch/models/thl/wallet/user_wallet.py index cbbc6d1..0a66fc2 100644 --- a/generalresearch/models/thl/wallet/user_wallet.py +++ b/generalresearch/models/thl/wallet/user_wallet.py @@ -1,11 +1,17 @@ from __future__ import annotations import logging +from typing import Literal from pydantic import BaseModel, ConfigDict, Field, NonNegativeInt +from generalresearch.models.custom_types import UUIDStr from generalresearch.models.legacy.api_status import StatusResponse -from generalresearch.models.thl.payout_format import PayoutFormatField, PayoutFormatType +from generalresearch.models.thl.ledger import AccountType +from generalresearch.models.thl.payout_format import ( + PayoutFormatField, + PayoutFormatType, +) logger = logging.getLogger() @@ -37,3 +43,40 @@ class UserWalletBalance(BaseModel): class UserWalletBalanceResponse(StatusResponse): wallet: UserWalletBalance = Field() + + +class UserLedgerWallet(UserWalletBalance): + """A user-owned ledger account exposed by the wallets endpoint.""" + + payout_format: PayoutFormatType | None = Field(default=None) + amount_string: str | None = Field(default=None) + redeemable_amount_string: str | None = Field(default=None) + + account_uuid: UUIDStr = Field( + description="A unique identifier for this Ledger Account", + examples=["c3c3566b5b1b4961b63a5670a2dc923d"], + ) + account_type: Literal[ + AccountType.USER_WALLET, + AccountType.USER_ATTEMPT_CREDIT, + ] + currency: str = Field(default="USD", max_length=32) + display_name: str = Field( + max_length=64, + description="Human-readable description of the Ledger Account", + ) + + +class UserDisplayedWalletBalance(BaseModel): + """Combined user-visible balance for one ledger currency.""" + + currency: str = Field(max_length=32) + amount: int = Field( + description="Displayed balance in the currency's smallest ledger unit." + ) + amount_string: str | None = None + + +class UserLedgerWallets(BaseModel): + wallets: list[UserLedgerWallet] = Field(default_factory=list) + displayed_balances: list[UserDisplayedWalletBalance] = Field(default_factory=list) diff --git a/tests/models/thl/test_product_userwalletconfig.py b/tests/models/thl/test_product_userwalletconfig.py index b1bfa60..9fb9c73 100644 --- a/tests/models/thl/test_product_userwalletconfig.py +++ b/tests/models/thl/test_product_userwalletconfig.py @@ -23,6 +23,7 @@ class TestProductUserWalletConfig: # Check the defaults assert not instance.enabled + assert instance.balance_type == "wallet_balance" assert not instance.amt assert isinstance(instance.supported_payout_types, set) @@ -30,6 +31,11 @@ class TestProductUserWalletConfig: assert instance.min_cashout is None + def test_redeemable_balance_type(self): + instance = UserWalletConfig(balance_type="redeemable_balance") + + assert instance.balance_type == "redeemable_balance" + def test_model_dump(self): instance = UserWalletConfig() |