implement cashout method popularity_rank

3 files changed, 157 insertions, 2 deletions

diff --git a/generalresearch/managers/thl/wallet/__init__.py b/generalresearch/managers/thl/wallet/__init__.py
index cc799d6..1e80397 100644
--- a/generalresearch/managers/thl/wallet/__init__.py
+++ b/generalresearch/managers/thl/wallet/__init__.py
@@ -14,13 +14,13 @@ if TYPE_CHECKING:
     from generalresearch.managers.thl.ledger_manager.thl_ledger import (
         ThlLedgerManager,
     )
-    from generalresearch.managers.thl.payout import UserPayoutEventManager
     from generalresearch.managers.thl.paypal import PayPalPayoutManager
     from generalresearch.managers.thl.tango_api import TangoClient
     from generalresearch.managers.thl.user_manager.user_manager import (
         UserManager,
     )
     from generalresearch.managers.thl.userhealth import UserIpHistoryManager
+    from generalresearch.managers.thl.wallet.user_payout import UserPayoutEventManager
     from generalresearch.models.thl.payout import UserPayoutEvent
     from generalresearch.models.thl.wallet.cashout_method import (
         CashMailOrderData,
@@ -34,7 +34,6 @@ def manage_pending_cashout(
     user_ip_history_manager: UserIpHistoryManager,
     user_manager: UserManager,
     ledger_manager: ThlLedgerManager,
-    geoip_info_manager: GeoIpInfoManager,
     order_data: dict[str, Any] | CashMailOrderData | None = None,
     tango_client: TangoClient | None = None,
     paypal_client: PayPalPayoutManager | None = None,
diff --git a/generalresearch/managers/thl/wallet/user_payout.py b/generalresearch/managers/thl/wallet/user_payout.py
index 99f80ef..2e9e7df 100644
--- a/generalresearch/managers/thl/wallet/user_payout.py
+++ b/generalresearch/managers/thl/wallet/user_payout.py
@@ -1,6 +1,8 @@
+import logging
 import uuid
 from collections.abc import Collection
 from datetime import UTC, datetime
+from threading import Thread
 from typing import Any
 from uuid import uuid4
 
@@ -26,14 +28,147 @@ from generalresearch.models.thl.wallet.cashout_method import (
     CashMailCashoutMethodRequestData,
     CashMailOrderData,
     CashoutMethod,
+    CashoutMethodOut,
     CashoutRequestDetail,
     PaypalCashoutMethodRequestData,
     TangoCashoutMethodRequestData,
 )
 from generalresearch.models.thl.wallet.definitions import PayoutType
 
+logger = logging.getLogger(__name__)
+
 
 class UserPayoutEventManager(PayoutEventManager):
+    CASHOUT_METHOD_REDEMPTION_COUNT_KEY = "cashout-method:redemption-count:v1"
+
+    def update(
+        self,
+        payout_event: UserPayoutEvent,
+        status: PayoutStatus,
+        ext_ref_id: str | None = None,
+        order_data: dict[str, Any] | None = None,
+    ) -> None:
+        was_complete = payout_event.status == PayoutStatus.COMPLETE
+        super().update(
+            payout_event=payout_event,
+            status=status,
+            ext_ref_id=ext_ref_id,
+            order_data=order_data,
+        )
+        if status != PayoutStatus.COMPLETE or was_complete:
+            return
+
+        assert payout_event.cashout_method_uuid is not None
+        key = CashoutMethod.make_redemption_key(
+            payout_event.payout_type,
+            payout_event.cashout_method_uuid,
+        )
+        try:
+            redis_key = self.CASHOUT_METHOD_REDEMPTION_COUNT_KEY
+            if self.redis_client.exists(redis_key):
+                self.redis_client.zincrby(redis_key, 1, key)
+        except Exception:
+            logger.exception(
+                "Unable to increment cashout-method redemption count",
+                extra={"payout_event_uuid": payout_event.uuid},
+            )
+
+    def _query_cashout_method_redemption_counts(self) -> dict[str, int]:
+        rows = self.pg_config.execute_sql_query(
+            query="""
+                SELECT CASE
+                           WHEN ac.user_id IS NOT NULL
+                               THEN 'provider:' || ac.provider
+                           ELSE 'id:' || REPLACE(ep.cashout_method_uuid::text, '-', '')
+                           END AS key,
+                       COUNT(*) AS redemption_count
+                FROM event_payout AS ep
+                JOIN accounting_cashoutmethod AS ac
+                    ON ac.id = ep.cashout_method_uuid
+                JOIN ledger_account AS la
+                    ON la.uuid = ep.debit_account_uuid
+                WHERE ac.provider != 'AMT'
+                  AND la.reference_type = 'user'
+                  AND ep.status = 'COMPLETE'
+                GROUP BY key
+            """
+        )
+        return {row["key"]: int(row["redemption_count"]) for row in rows}
+
+    def rebuild_cashout_method_redemption_counts(self) -> dict[str, int]:
+        counts = self._query_cashout_method_redemption_counts()
+        redis_key = self.CASHOUT_METHOD_REDEMPTION_COUNT_KEY
+        with self.redis_client.pipeline() as pipeline:
+            pipeline.delete(redis_key)
+            pipeline.zadd(redis_key, counts)
+            pipeline.execute()
+        return counts
+
+    def _rebuild_cashout_method_redemption_counts_if_missing(self) -> None:
+        redis_key = self.CASHOUT_METHOD_REDEMPTION_COUNT_KEY
+        lock = self.redis_client.lock(
+            f"{redis_key}:rebuild-lock",
+            timeout=60,
+            blocking_timeout=0,
+        )
+        if not lock.acquire():
+            return
+        try:
+            if not self.redis_client.exists(redis_key):
+                self.rebuild_cashout_method_redemption_counts()
+        except Exception:
+            logger.exception("Unable to rebuild cashout-method redemption counts")
+        finally:
+            lock.release()
+
+    def get_cashout_method_redemption_counts(
+        self, cashout_methods: Collection[CashoutMethod | CashoutMethodOut]
+    ) -> dict[str, int]:
+        methods = list(cashout_methods)
+        redis_key = self.CASHOUT_METHOD_REDEMPTION_COUNT_KEY
+        try:
+            if not self.redis_client.exists(redis_key):
+                Thread(
+                    target=self._rebuild_cashout_method_redemption_counts_if_missing,
+                    daemon=True,
+                ).start()
+                return {method.id: 0 for method in methods}
+
+            with self.redis_client.pipeline() as pipeline:
+                for method in methods:
+                    pipeline.zscore(
+                        redis_key,
+                        CashoutMethod.make_redemption_key(method.type, method.id),
+                    )
+                scores = pipeline.execute()
+            return {
+                method.id: int(score or 0)
+                for method, score in zip(methods, scores, strict=True)
+            }
+        except Exception:
+            logger.exception("Unable to read cashout-method redemption counts")
+            return {method.id: 0 for method in methods}
+
+    def apply_cashout_method_popularity(
+        self,
+        cashout_methods: list[CashoutMethodOut],
+    ) -> list[CashoutMethodOut]:
+        redemption_counts = self.get_cashout_method_redemption_counts(cashout_methods)
+        ranked_methods = sorted(
+            cashout_methods,
+            key=lambda method: (
+                -redemption_counts[method.id],
+                method.name.casefold(),
+                method.id,
+            ),
+        )
+        popularity_ranks = {
+            method.id: rank for rank, method in enumerate(ranked_methods, start=1)
+        }
+        for method in cashout_methods:
+            method.popularity_rank = popularity_ranks[method.id]
+        return ranked_methods
+
     def get_by_uuid(self, pe_uuid: UUIDStr) -> UserPayoutEvent:
 
         res = self.pg_config.execute_sql_query(
diff --git a/generalresearch/models/thl/wallet/cashout_method.py b/generalresearch/models/thl/wallet/cashout_method.py
index 518aa2f..6357f9f 100644
--- a/generalresearch/models/thl/wallet/cashout_method.py
+++ b/generalresearch/models/thl/wallet/cashout_method.py
@@ -178,6 +178,19 @@ class CashoutMethod(CashoutMethodBase):
             )
         return self
 
+    @property
+    def redemption_key(self) -> str:
+        return self.make_redemption_key(self.type, self.id)
+
+    @staticmethod
+    def make_redemption_key(payout_type: PayoutType, cashout_method_id: str) -> str:
+        # The format of the key is dependent on the cashout method provider.
+        # e.g., Paypal has individual cashout methods per user, so they are grouped.
+        # "provider:PAYPAL", "provider:CASH_IN_MAIL", "id:941d489c3ce04eb39a0ddb7f8f75db74"
+        if payout_type in {PayoutType.PAYPAL, PayoutType.CASH_IN_MAIL}:
+            return f"provider:{payout_type.value}"
+        return f"id:{cashout_method_id}"
+
 
 class CashoutMethodOut(CashoutMethodBase):
     product_id: UUIDStr | None = Field(
@@ -201,6 +214,11 @@ class CashoutMethodOut(CashoutMethodBase):
         "to meet this cashout method's minimum value. None when not evaluated for "
         "a user.",
     )
+    popularity_rank: PositiveInt | None = Field(
+        default=None,
+        description="The method's popularity rank among the available methods. "
+        "One is the most popular.",
+    )
 
     @classmethod
     def from_cashout_method(
@@ -208,8 +226,11 @@ class CashoutMethodOut(CashoutMethodBase):
         cm: CashoutMethod,
         *,
         cashout_balance: int | None = None,
+        popularity_rank: int | None = None,
     ) -> Self:
         d = cm.model_dump()
+        if popularity_rank is not None:
+            d["popularity_rank"] = popularity_rank
         if cashout_balance is not None and cm.min_value_usd is not None:
             d["can_redeem"] = cashout_balance >= int(cm.min_value_usd)
         if cm.user: