feat: grant signup bonus points

apps/billing/tests.py

@@ -24,6 +24,7 @@ from apps.billing.models import (
     PointsLedger,
     PricingRule,
     RechargeOrder,
+    SignupBonusGrant,
 )
 from apps.billing.pricing import (
     NoPricingRuleError,
@@ -45,6 +46,7 @@ from apps.billing.services import (
     adjust_wallet_points,
     apply_recharge_payment,
     create_recharge_order,
+    grant_signup_bonus,
     mark_call_success,
     precharge_call,
     query_and_apply_recharge_payment,
@@ -273,6 +275,7 @@ class BillingCoreModelTests(TestCase):
         self.assertIn(PointsLedger, admin.site._registry)
         self.assertIn(CallRecord, admin.site._registry)
         self.assertIn(RechargeOrder, admin.site._registry)
+        self.assertIn(SignupBonusGrant, admin.site._registry)
 
 
 @override_settings(AI_KEY_ENCRYPTION_KEY=TEST_ENCRYPTION_KEY)
@@ -484,6 +487,50 @@ class BillingServiceTests(TestCase):
         self.assertEqual(order.exchange_rate, Decimal("12.5000"))
         self.assertEqual(order.points_granted, 111)
 
+    def test_grant_signup_bonus_creates_wallet_ledger_and_is_idempotent(self):
+        suffix = uuid.uuid4().hex[:8]
+        user = get_user_model().objects.create_user(
+            username=f"signup-bonus-{suffix}",
+            email=f"signup-bonus-{suffix}@example.com",
+            password="password",
+        )
+
+        first = grant_signup_bonus(user=user)
+        second = grant_signup_bonus(user=user)
+
+        wallet = UserWallet.objects.get(user=user)
+        self.assertTrue(first.granted)
+        self.assertFalse(second.granted)
+        self.assertEqual(first.points_granted, 100)
+        self.assertEqual(second.points_granted, 0)
+        self.assertEqual(first.balance_after, 100)
+        self.assertEqual(second.balance_after, 100)
+        self.assertEqual(wallet.points_balance, 100)
+        self.assertEqual(SignupBonusGrant.objects.filter(user=user).count(), 1)
+        ledger = PointsLedger.objects.get(
+            user=user,
+            change_type=PointsLedger.ChangeType.SIGNUP_BONUS,
+        )
+        self.assertEqual(ledger.points_delta, 100)
+        self.assertEqual(ledger.balance_after, 100)
+        self.assertEqual(ledger.reason, "new_user_registration")
+
+    def test_grant_signup_bonus_adds_to_existing_wallet_balance(self):
+        suffix = uuid.uuid4().hex[:8]
+        user = get_user_model().objects.create_user(
+            username=f"signup-existing-wallet-{suffix}",
+            email=f"signup-existing-wallet-{suffix}@example.com",
+            password="password",
+        )
+        UserWallet.objects.create(user=user, points_balance=30)
+
+        result = grant_signup_bonus(user=user)
+
+        wallet = UserWallet.objects.get(user=user)
+        self.assertTrue(result.granted)
+        self.assertEqual(wallet.points_balance, 130)
+        self.assertEqual(result.balance_after, 130)
+
     def test_create_recharge_order_marks_order_failed_when_gateway_fails(self):
         self.create_exchange_rate(points_per_unit="10.0000")
 
@@ -957,3 +1004,69 @@ class ConcurrentDebitTests(TransactionTestCase):
                 balance_after__lt=0,
             ).exists()
         )
+
+
+class ConcurrentSignupBonusTests(TransactionTestCase):
+    def setUp(self):
+        self.addCleanup(connections.close_all)
+        suffix = uuid.uuid4().hex[:8]
+        self.user = get_user_model().objects.create_user(
+            username=f"concurrent-signup-{suffix}",
+            email=f"concurrent-signup-{suffix}@example.com",
+            password="password",
+        )
+        UserWallet.objects.create(user=self.user, points_balance=0)
+
+    def test_concurrent_signup_bonus_is_granted_once(self):
+        worker_count = 2
+        barrier = threading.Barrier(worker_count)
+        results = []
+        errors = []
+        results_lock = threading.Lock()
+
+        def worker():
+            connections.close_all()
+            try:
+                for attempt in range(3):
+                    try:
+                        connections["default"].ensure_connection()
+                        break
+                    except OperationalError as exc:
+                        connections.close_all()
+                        if attempt == 2:
+                            with results_lock:
+                                errors.append(repr(exc))
+                            barrier.abort()
+                            return
+                        time.sleep(0.5)
+                barrier.wait(timeout=15)
+                result = grant_signup_bonus(user=self.user)
+                with results_lock:
+                    results.append((result.granted, result.balance_after))
+            except Exception as exc:  # pragma: no cover - surfaced through assertion below.
+                with results_lock:
+                    errors.append(repr(exc))
+            finally:
+                connections.close_all()
+
+        threads = [threading.Thread(target=worker) for _index in range(worker_count)]
+        for thread in threads:
+            thread.start()
+        for thread in threads:
+            thread.join(timeout=30)
+
+        self.assertFalse(any(thread.is_alive() for thread in threads), "worker thread timed out")
+        self.assertEqual(errors, [])
+        self.assertEqual(sum(1 for granted, _balance in results if granted), 1)
+        self.assertEqual(sum(1 for granted, _balance in results if not granted), 1)
+
+        wallet = UserWallet.objects.get(user=self.user)
+        self.assertEqual(wallet.points_balance, 100)
+        self.assertEqual(SignupBonusGrant.objects.filter(user=self.user).count(), 1)
+        self.assertEqual(
+            PointsLedger.objects.filter(
+                user=self.user,
+                change_type=PointsLedger.ChangeType.SIGNUP_BONUS,
+            ).count(),
+            1,
+        )