gavindiaz/PolyWeather
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add trial and referral subscription program

Browse Source
This commit is contained in:
2569718930@qq.com
2026-05-29 19:24:46 +08:00
parent f8f5035225
commit 522e35de7f
20 changed files with 1629 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/auth/supabase_entitlement.py
+601 -2
View File
@@ -1,15 +1,28 @@
from __future__ import annotations
import os
import secrets
import threading
import time
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Dict, List, Optional
from datetime import datetime, timedelta, timezone
from typing import Any, Dict, List, Optional
import requests
from loguru import logger
from src.database.db_manager import DBManager
SIGNUP_TRIAL_PLAN_CODE = "signup_trial_3d"
SIGNUP_TRIAL_SOURCE = "signup_trial"
SIGNUP_TRIAL_DAYS = 3
REFERRAL_REWARD_DAYS = 3
REFERRAL_MONTHLY_REWARD_LIMIT = 10
REFERRAL_MONTHLY_DAY_LIMIT = 30
REFERRAL_DISCOUNT_USDC = "3"
REFERRAL_MONTHLY_DISCOUNTED_AMOUNT_USDC = "26.9"
def _env_bool(name: str, default: bool = False) -> bool:
raw = os.getenv(name)
@@ -121,9 +134,595 @@ class SupabaseEntitlementService:
"Accept": "application/json",
}
def _service_rest_headers(self, prefer: Optional[str] = None) -> Dict[str, str]:
headers = self._request_headers_for_service_role()
headers["Content-Type"] = "application/json"
if prefer:
headers["Prefer"] = prefer
return headers
def _rest(
self,
method: str,
table: str,
*,
params: Optional[Dict[str, Any]] = None,
payload: Optional[Any] = None,
prefer: Optional[str] = None,
allowed_status: Optional[List[int]] = None,
) -> Any:
if not self.supabase_url or not self.service_role_key:
raise RuntimeError("supabase service role is not configured")
status_ok = allowed_status or [200, 201, 204]
response = requests.request(
method=method.upper(),
url=f"{self.supabase_url}/rest/v1/{table}",
headers=self._service_rest_headers(prefer=prefer),
params=params,
json=payload,
timeout=self.timeout_sec,
)
if response.status_code not in status_ok:
detail = response.text[:350] if response.text else response.reason
raise RuntimeError(
f"supabase {method.upper()} {table} failed: "
f"{response.status_code} {detail}"
)
if not response.content:
return None
try:
return response.json()
except Exception:
return None
def _admin_user_endpoint(self, user_id: str) -> str:
return f"{self.supabase_url}/auth/v1/admin/users/{user_id}"
@staticmethod
def _to_iso(dt: datetime) -> str:
return dt.astimezone(timezone.utc).isoformat()
@staticmethod
def _normalize_email(value: Optional[str]) -> str:
return str(value or "").strip().lower()
@staticmethod
def _is_trial_subscription_row(row: Optional[Dict[str, object]]) -> bool:
if not isinstance(row, dict):
return False
plan_code = str(row.get("plan_code") or "").strip().lower()
source = str(row.get("source") or "").strip().lower()
return "trial" in plan_code or "trial" in source
@staticmethod
def _is_paid_subscription_row(row: Optional[Dict[str, object]]) -> bool:
if not isinstance(row, dict):
return False
if SupabaseEntitlementService._is_trial_subscription_row(row):
return False
source = str(row.get("source") or "").strip().lower()
if "referral_reward" in source:
return False
return "payment" in source or source in {"payment_contract", "payment_manual"}
def _telegram_user_id_for(self, user_id: str) -> Optional[int]:
try:
linked = DBManager().get_user_by_supabase_user_id(user_id)
if not isinstance(linked, dict):
return None
telegram_id = int(linked.get("telegram_id") or 0)
return telegram_id or None
except Exception:
return None
def _active_wallet_addresses_for(self, user_id: str) -> List[str]:
try:
rows = self._rest(
"GET",
"user_wallets",
params={
"select": "address",
"user_id": f"eq.{user_id}",
"status": "eq.active",
"limit": "50",
},
allowed_status=[200],
)
except Exception:
return []
out: List[str] = []
if isinstance(rows, list):
for row in rows:
if not isinstance(row, dict):
continue
address = str(row.get("address") or "").strip().lower()
if address and address not in out:
out.append(address)
return out
def _trial_claim_exists(
self,
*,
user_id: str,
email: str,
telegram_user_id: Optional[int],
wallet_addresses: List[str],
) -> bool:
checks = [f"user_id.eq.{user_id}"]
if email:
checks.append(f"email.eq.{email}")
if telegram_user_id:
checks.append(f"telegram_user_id.eq.{telegram_user_id}")
try:
rows = self._rest(
"GET",
"trial_claims",
params={
"select": "id",
"or": f"({','.join(checks)})",
"limit": "1",
},
allowed_status=[200],
)
if isinstance(rows, list) and rows:
return True
except Exception:
raise
if not wallet_addresses:
return False
try:
wallet_rows = self._rest(
"GET",
"trial_claim_wallets",
params={
"select": "id",
"wallet_address": f"in.({','.join(wallet_addresses)})",
"limit": "1",
},
allowed_status=[200],
)
return bool(isinstance(wallet_rows, list) and wallet_rows)
except Exception:
return False
def ensure_signup_trial(self, user_id: str, email: Optional[str] = None) -> Dict[str, object]:
user_key = str(user_id or "").strip()
if not user_key:
return {"created": False, "reason": "missing_user_id"}
if not _env_bool("POLYWEATHER_SIGNUP_TRIAL_ENABLED", True):
return {"created": False, "reason": "disabled"}
if not self.supabase_url or not self.service_role_key:
return {"created": False, "reason": "supabase_not_configured"}
normalized_email = self._normalize_email(email)
try:
telegram_user_id = self._telegram_user_id_for(user_key)
wallet_addresses = self._active_wallet_addresses_for(user_key)
if self._trial_claim_exists(
user_id=user_key,
email=normalized_email,
telegram_user_id=telegram_user_id,
wallet_addresses=wallet_addresses,
):
return {"created": False, "reason": "already_claimed"}
now = datetime.now(timezone.utc)
expires = now + timedelta(days=SIGNUP_TRIAL_DAYS)
claim_payload = {
"user_id": user_key,
"email": normalized_email,
"telegram_user_id": telegram_user_id,
"primary_wallet_address": wallet_addresses[0] if wallet_addresses else None,
"claimed_at": self._to_iso(now),
"metadata": {"wallet_addresses": wallet_addresses},
}
claim_rows = self._rest(
"POST",
"trial_claims",
payload=claim_payload,
prefer="return=representation",
allowed_status=[200, 201],
)
claim_id = None
if isinstance(claim_rows, list) and claim_rows and isinstance(claim_rows[0], dict):
claim_id = claim_rows[0].get("id")
if wallet_addresses and claim_id is not None:
self._rest(
"POST",
"trial_claim_wallets",
payload=[
{
"trial_claim_id": claim_id,
"wallet_address": address,
"created_at": self._to_iso(now),
}
for address in wallet_addresses
],
prefer="return=minimal",
allowed_status=[201],
)
subscription_payload = {
"user_id": user_key,
"plan_code": SIGNUP_TRIAL_PLAN_CODE,
"status": "active",
"starts_at": self._to_iso(now),
"expires_at": self._to_iso(expires),
"source": SIGNUP_TRIAL_SOURCE,
"created_at": self._to_iso(now),
"updated_at": self._to_iso(now),
}
self._rest(
"POST",
"subscriptions",
payload=subscription_payload,
prefer="return=minimal",
allowed_status=[201],
)
self._rest(
"POST",
"entitlement_events",
payload={
"user_id": user_key,
"action": "signup_trial_granted",
"reason": "first_auth",
"actor": "supabase_auth",
"payload": {
"plan_code": SIGNUP_TRIAL_PLAN_CODE,
"expires_at": self._to_iso(expires),
},
"created_at": self._to_iso(now),
},
prefer="return=minimal",
allowed_status=[201],
)
self.invalidate_subscription_cache(user_key)
return {
"created": True,
"plan_code": SIGNUP_TRIAL_PLAN_CODE,
"expires_at": self._to_iso(expires),
}
except Exception as exc:
logger.warning("signup trial grant failed user_id={}: {}", user_key, exc)
return {"created": False, "reason": "error"}
def has_paid_subscription(self, user_id: str) -> bool:
user_key = str(user_id or "").strip()
if not user_key:
return False
try:
rows = self._rest(
"GET",
"subscriptions",
params={
"select": "plan_code,source,status,starts_at,expires_at",
"user_id": f"eq.{user_key}",
"limit": "100",
},
allowed_status=[200],
)
except Exception:
return False
if not isinstance(rows, list):
return False
return any(self._is_paid_subscription_row(row) for row in rows if isinstance(row, dict))
@staticmethod
def _normalize_referral_code(value: Optional[str]) -> str:
return "".join(str(value or "").strip().upper().split())
def ensure_referral_code(self, user_id: str) -> Optional[Dict[str, object]]:
user_key = str(user_id or "").strip()
if not user_key or not self.service_role_key:
return None
try:
rows = self._rest(
"GET",
"referral_codes",
params={
"select": "code,status,created_at",
"user_id": f"eq.{user_key}",
"status": "eq.active",
"limit": "1",
},
allowed_status=[200],
)
if isinstance(rows, list) and rows and isinstance(rows[0], dict):
return rows[0]
now = datetime.now(timezone.utc)
for _ in range(5):
code = f"PW{secrets.token_hex(4).upper()}"
try:
created = self._rest(
"POST",
"referral_codes",
payload={
"user_id": user_key,
"code": code,
"status": "active",
"created_at": self._to_iso(now),
"updated_at": self._to_iso(now),
},
prefer="return=representation",
allowed_status=[200, 201],
)
if isinstance(created, list) and created and isinstance(created[0], dict):
return created[0]
return {"code": code, "status": "active"}
except Exception:
continue
except Exception as exc:
logger.warning("referral code ensure failed user_id={}: {}", user_key, exc)
return None
def get_pending_referral_attribution(self, user_id: str) -> Optional[Dict[str, object]]:
user_key = str(user_id or "").strip()
if not user_key:
return None
try:
rows = self._rest(
"GET",
"referral_attributions",
params={
"select": "id,code,referrer_user_id,referred_user_id,status,created_at",
"referred_user_id": f"eq.{user_key}",
"status": "eq.pending",
"order": "created_at.desc",
"limit": "1",
},
allowed_status=[200],
)
except Exception:
return None
if isinstance(rows, list) and rows and isinstance(rows[0], dict):
return rows[0]
return None
def _current_month_reward_rows(self, referrer_user_id: str) -> List[Dict[str, object]]:
month_start = datetime.now(timezone.utc).replace(
day=1,
hour=0,
minute=0,
second=0,
microsecond=0,
)
try:
rows = self._rest(
"GET",
"referral_rewards",
params={
"select": "id,reward_days,created_at",
"referrer_user_id": f"eq.{referrer_user_id}",
"created_at": f"gte.{self._to_iso(month_start)}",
"limit": "100",
},
allowed_status=[200],
)
except Exception:
return []
return [row for row in rows if isinstance(row, dict)] if isinstance(rows, list) else []
def get_referral_summary(self, user_id: str) -> Optional[Dict[str, object]]:
user_key = str(user_id or "").strip()
if not user_key or not self.service_role_key:
return None
try:
code_row = self.ensure_referral_code(user_key) or {}
pending = self.get_pending_referral_attribution(user_key)
rewards = self._current_month_reward_rows(user_key)
reward_count = len(rewards)
reward_days = sum(int(row.get("reward_days") or 0) for row in rewards)
return {
"code": str(code_row.get("code") or ""),
"discount_usdc": REFERRAL_DISCOUNT_USDC,
"discounted_monthly_amount_usdc": REFERRAL_MONTHLY_DISCOUNTED_AMOUNT_USDC,
"reward_days": REFERRAL_REWARD_DAYS,
"monthly_reward_limit": REFERRAL_MONTHLY_REWARD_LIMIT,
"monthly_reward_days_limit": REFERRAL_MONTHLY_DAY_LIMIT,
"monthly_reward_count": reward_count,
"monthly_reward_days": min(reward_days, REFERRAL_MONTHLY_DAY_LIMIT),
"applied_code": str(pending.get("code") or "") if isinstance(pending, dict) else "",
"attribution_status": str(pending.get("status") or "") if isinstance(pending, dict) else "",
}
except Exception as exc:
logger.warning("referral summary failed user_id={}: {}", user_key, exc)
return None
def apply_referral_code(self, user_id: str, code: str) -> Dict[str, object]:
user_key = str(user_id or "").strip()
normalized_code = self._normalize_referral_code(code)
if not user_key:
raise ValueError("user_id required")
if len(normalized_code) < 3:
raise ValueError("invalid referral code")
if self.has_paid_subscription(user_key):
raise ValueError("referral code can only be used before first paid subscription")
rows = self._rest(
"GET",
"referral_codes",
params={
"select": "user_id,code,status",
"code": f"eq.{normalized_code}",
"status": "eq.active",
"limit": "1",
},
allowed_status=[200],
)
if not isinstance(rows, list) or not rows or not isinstance(rows[0], dict):
raise ValueError("referral code not found")
referrer_user_id = str(rows[0].get("user_id") or "").strip()
if not referrer_user_id or referrer_user_id == user_key:
raise ValueError("cannot use your own referral code")
existing = self.get_pending_referral_attribution(user_key)
if isinstance(existing, dict):
return {
"ok": True,
"already_applied": True,
"referral": self.get_referral_summary(user_key),
}
now = datetime.now(timezone.utc)
self._rest(
"POST",
"referral_attributions",
payload={
"referrer_user_id": referrer_user_id,
"referred_user_id": user_key,
"code": normalized_code,
"status": "pending",
"created_at": self._to_iso(now),
"updated_at": self._to_iso(now),
},
prefer="return=minimal",
allowed_status=[201],
)
return {
"ok": True,
"already_applied": False,
"referral": self.get_referral_summary(user_key),
}
def _subscription_extension_start(self, user_id: str) -> datetime:
now = datetime.now(timezone.utc)
try:
rows = self._rest(
"GET",
"subscriptions",
params={
"select": "starts_at,expires_at,plan_code,source",
"user_id": f"eq.{user_id}",
"status": "eq.active",
"expires_at": f"gt.{self._to_iso(now)}",
"order": "expires_at.desc",
"limit": "20",
},
allowed_status=[200],
)
except Exception:
return now
starts = now
if isinstance(rows, list):
for row in rows:
if not isinstance(row, dict) or self._is_trial_subscription_row(row):
continue
exp = self._parse_iso_datetime(str(row.get("expires_at") or ""))
starts_at = self._parse_iso_datetime(str(row.get("starts_at") or ""))
if exp and (starts_at is None or starts_at <= now) and exp > starts:
starts = exp
break
return starts
def settle_referral_reward(
self,
*,
referred_user_id: str,
payment_intent_id: str,
tx_hash: str,
) -> Dict[str, object]:
referred_key = str(referred_user_id or "").strip()
attribution = self.get_pending_referral_attribution(referred_key)
if not isinstance(attribution, dict):
return {"awarded": False, "reason": "no_pending_referral"}
referrer_key = str(attribution.get("referrer_user_id") or "").strip()
if not referrer_key or referrer_key == referred_key:
return {"awarded": False, "reason": "invalid_referrer"}
now = datetime.now(timezone.utc)
monthly_rewards = self._current_month_reward_rows(referrer_key)
if len(monthly_rewards) >= REFERRAL_MONTHLY_REWARD_LIMIT:
self._rest(
"PATCH",
"referral_attributions",
params={"id": f"eq.{attribution.get('id')}"},
payload={
"status": "capped",
"updated_at": self._to_iso(now),
"converted_payment_intent_id": payment_intent_id,
"converted_tx_hash": tx_hash,
},
prefer="return=minimal",
allowed_status=[204],
)
return {"awarded": False, "reason": "monthly_cap_reached"}
starts = self._subscription_extension_start(referrer_key)
expires = starts + timedelta(days=REFERRAL_REWARD_DAYS)
subscription_payload = {
"user_id": referrer_key,
"plan_code": "pro_monthly",
"status": "active",
"starts_at": self._to_iso(starts),
"expires_at": self._to_iso(expires),
"source": "referral_reward",
"created_at": self._to_iso(now),
"updated_at": self._to_iso(now),
}
self._rest(
"POST",
"subscriptions",
payload=subscription_payload,
prefer="return=minimal",
allowed_status=[201],
)
self._rest(
"POST",
"referral_rewards",
payload={
"referral_attribution_id": attribution.get("id"),
"referrer_user_id": referrer_key,
"referred_user_id": referred_key,
"payment_intent_id": payment_intent_id,
"tx_hash": tx_hash,
"reward_days": REFERRAL_REWARD_DAYS,
"created_at": self._to_iso(now),
},
prefer="return=minimal",
allowed_status=[201],
)
self._rest(
"PATCH",
"referral_attributions",
params={"id": f"eq.{attribution.get('id')}"},
payload={
"status": "converted",
"converted_payment_intent_id": payment_intent_id,
"converted_tx_hash": tx_hash,
"converted_at": self._to_iso(now),
"updated_at": self._to_iso(now),
},
prefer="return=minimal",
allowed_status=[204],
)
self._rest(
"POST",
"entitlement_events",
payload={
"user_id": referrer_key,
"action": "referral_reward_granted",
"reason": "referred_user_paid",
"actor": "payment_contract_checkout",
"payload": {
"referred_user_id": referred_key,
"payment_intent_id": payment_intent_id,
"tx_hash": tx_hash,
"reward_days": REFERRAL_REWARD_DAYS,
},
"created_at": self._to_iso(now),
},
prefer="return=minimal",
allowed_status=[201],
)
self.invalidate_subscription_cache(referrer_key)
return {
"awarded": True,
"reward_days": REFERRAL_REWARD_DAYS,
"referrer_user_id": referrer_key,
"subscription": subscription_payload,
}
def get_identity(self, access_token: str) -> Optional[SupabaseIdentity]:
if not access_token:
return None
src/payments/contract_checkout.py
+123 -11
View File
@@ -13,6 +13,7 @@ from typing import Any, Dict, List, Optional
import requests
from eth_account import Account
from eth_account.messages import encode_defunct
from loguru import logger
from web3 import Web3
from src.auth.supabase_entitlement import SUPABASE_ENTITLEMENT
@@ -90,9 +91,12 @@ ERC20_TRANSFER_EVENT_ABI = {
}
DEFAULT_PLAN_CATALOG: Dict[str, Dict[str, Any]] = {
"pro_monthly": {"plan_id": 101, "amount_usdc": "10", "duration_days": 30},
"pro_monthly": {"plan_id": 101, "amount_usdc": "29.9", "duration_days": 30},
"pro_quarterly": {"plan_id": 102, "amount_usdc": "79.9", "duration_days": 90},
}
REFERRAL_FIRST_MONTH_DISCOUNT_USDC = Decimal("3")
def _env_bool(name: str, default: bool = False) -> bool:
raw = os.getenv(name)
@@ -195,13 +199,13 @@ def _parse_plan_catalog(raw: str) -> Dict[str, Dict[str, Any]]:
def _parse_allowed_plan_codes(raw: str) -> List[str]:
text = str(raw or "").strip()
if not text:
return ["pro_monthly"]
return ["pro_monthly", "pro_quarterly"]
out: List[str] = []
for part in text.split(","):
code = str(part or "").strip().lower()
if code and code not in out:
out.append(code)
return out or ["pro_monthly"]
return out or ["pro_monthly", "pro_quarterly"]
@dataclass
@@ -363,6 +367,10 @@ class PaymentContractCheckoutService:
self.notify_telegram = _env_bool(
"POLYWEATHER_PAYMENT_TELEGRAM_NOTIFY_ENABLED", True
)
self.telegram_payment_pricing_enabled = _env_bool(
"POLYWEATHER_PAYMENT_TELEGRAM_PRICING_ENABLED",
False,
)
self.points_enabled = _env_bool("POLYWEATHER_PAYMENT_POINTS_ENABLED", True)
self.points_per_usdc = max(
1, _env_int("POLYWEATHER_PAYMENT_POINTS_PER_USDC", 500)
@@ -1650,6 +1658,55 @@ class PaymentContractCheckoutService:
out["telegram_pricing"] = price_payload
return out
def _get_pending_referral_attribution(self, user_id: str) -> Optional[Dict[str, Any]]:
try:
row = SUPABASE_ENTITLEMENT.get_pending_referral_attribution(user_id)
return dict(row) if isinstance(row, dict) else None
except Exception:
return None
def _has_prior_paid_subscription(self, user_id: str) -> bool:
try:
return bool(SUPABASE_ENTITLEMENT.has_paid_subscription(user_id))
except Exception:
return False
def _apply_referral_pricing(
self,
user_id: str,
plan: Dict[str, Any],
) -> Dict[str, Any]:
out = dict(plan)
attribution = self._get_pending_referral_attribution(user_id)
if not attribution or self._has_prior_paid_subscription(user_id):
return out
out["referral_attribution"] = {
"id": attribution.get("id"),
"code": str(attribution.get("code") or "").strip().upper(),
"referrer_user_id": str(attribution.get("referrer_user_id") or "").strip(),
"referred_user_id": user_id,
}
if str(out.get("plan_code") or "").strip().lower() != "pro_monthly":
return out
base_amount = _parse_decimal(out.get("amount_usdc_decimal"), Decimal("0"))
discount = min(REFERRAL_FIRST_MONTH_DISCOUNT_USDC, base_amount)
discounted = base_amount - discount
if discount <= 0 or discounted <= 0:
return out
out["amount_before_discount_usdc_decimal"] = base_amount
out["amount_usdc_decimal"] = discounted
out["amount_usdc"] = _format_decimal(discounted)
out["referral_discount"] = {
"discount_usdc": _format_decimal(discount),
"amount_before_discount_usdc": _format_decimal(base_amount),
"amount_after_discount_usdc": _format_decimal(discounted),
"reason": "first_month_referral",
}
return out
def _build_tx_payload(self, intent: PaymentIntentRecord) -> Dict[str, Any]:
contract = self._get_contract(intent.receiver_address, intent.chain_id)
tx_data = contract.encode_abi(
@@ -1687,10 +1744,10 @@ class PaymentContractCheckoutService:
points_to_consume: Optional[int] = None,
) -> Dict[str, Any]:
self._ensure_enabled()
plan = self._apply_telegram_group_pricing(
user_id,
self._select_plan(plan_code),
)
selected_plan = self._select_plan(plan_code)
if self.telegram_payment_pricing_enabled:
selected_plan = self._apply_telegram_group_pricing(user_id, selected_plan)
plan = self._apply_referral_pricing(user_id, selected_plan)
selected_token = self._resolve_supported_token(token_address, chain_id)
selected_chain_id = int(selected_token.chain_id)
mode = str(payment_mode or "strict").strip().lower()
@@ -1728,6 +1785,10 @@ class PaymentContractCheckoutService:
elif target_wallet:
self._require_user_wallet(user_id, target_wallet)
plan_amount_usdc = plan["amount_usdc_decimal"]
amount_before_discount_usdc = plan.get(
"amount_before_discount_usdc_decimal",
plan_amount_usdc,
)
redemption = self._build_points_redemption(
user_id=user_id,
plan_amount_usdc=plan_amount_usdc,
@@ -1748,13 +1809,17 @@ class PaymentContractCheckoutService:
combined_metadata["chain_name"] = selected_token.chain_name
if isinstance(plan.get("telegram_pricing"), dict):
combined_metadata["telegram_pricing"] = plan["telegram_pricing"]
if isinstance(plan.get("referral_attribution"), dict):
combined_metadata["referral_attribution"] = plan["referral_attribution"]
if isinstance(plan.get("referral_discount"), dict):
combined_metadata["referral_discount"] = plan["referral_discount"]
receiver_address = (
selected_token.direct_receiver_address
if mode == "direct"
else selected_token.receiver_contract
)
combined_metadata["amount_before_discount_usdc"] = _format_decimal(
plan_amount_usdc
amount_before_discount_usdc
)
combined_metadata["amount_after_discount_usdc"] = _format_decimal(
final_amount_usdc
@@ -1813,7 +1878,9 @@ class PaymentContractCheckoutService:
"plan_code": plan["plan_code"],
"plan_id": plan["plan_id"],
"duration_days": plan["duration_days"],
"amount_before_discount_usdc": _format_decimal(plan_amount_usdc),
"amount_before_discount_usdc": _format_decimal(
amount_before_discount_usdc
),
"amount_after_discount_usdc": _format_decimal(final_amount_usdc),
},
"token": {
@@ -2487,13 +2554,14 @@ class PaymentContractCheckoutService:
duration_days: int,
tx_hash: str,
payload: Dict[str, Any],
source: str = "payment",
) -> Dict[str, Any]:
now = _now_utc()
latest_rows = self._rest(
"GET",
"subscriptions",
params={
"select": "starts_at,expires_at",
"select": "starts_at,expires_at,plan_code,source",
"user_id": f"eq.{user_id}",
"status": "eq.active",
"order": "expires_at.desc",
@@ -2507,6 +2575,8 @@ class PaymentContractCheckoutService:
for row in latest_rows:
if not isinstance(row, dict):
continue
if self._subscription_row_is_trial(row):
continue
try:
starts_at = datetime.fromisoformat(
str(row.get("starts_at") or "").replace("Z", "+00:00")
@@ -2540,7 +2610,7 @@ class PaymentContractCheckoutService:
"status": "active",
"starts_at": _to_iso(starts),
"expires_at": _to_iso(expires),
"source": "payment_contract",
"source": str(source or "payment").strip() or "payment",
"created_at": _to_iso(now),
"updated_at": _to_iso(now),
}
@@ -2602,6 +2672,31 @@ class PaymentContractCheckoutService:
source = str(row.get("source") or "").strip().lower()
return "trial" in plan_code or "trial" in source
def _settle_referral_reward_for_intent(
self,
user_id: str,
intent: PaymentIntentRecord,
tx_hash: str,
) -> Dict[str, Any]:
metadata = dict(intent.metadata or {})
if not isinstance(metadata.get("referral_attribution"), dict):
return {}
try:
result = SUPABASE_ENTITLEMENT.settle_referral_reward(
referred_user_id=user_id,
payment_intent_id=intent.intent_id,
tx_hash=tx_hash,
)
return dict(result) if isinstance(result, dict) else {}
except Exception as exc:
logger.warning(
"referral reward settlement failed user_id={} intent_id={}: {}",
user_id,
intent.intent_id,
exc,
)
return {"awarded": False, "reason": "settlement_error"}
def _ensure_confirm_side_effects(
self,
user_id: str,
@@ -2624,9 +2719,15 @@ class PaymentContractCheckoutService:
},
)
subscription_row = self._ensure_confirmed_subscription(user_id, intent, tx_hash)
referral_reward = self._settle_referral_reward_for_intent(
user_id,
intent,
tx_hash,
)
return {
"payment": payment_row,
"subscription": subscription_row,
"referral_reward": referral_reward,
}
def _attempt_confirm_repair(
@@ -2779,6 +2880,7 @@ class PaymentContractCheckoutService:
"already_confirmed": True,
"payment": repaired.get("payment"),
"subscription": repaired.get("subscription"),
"referral_reward": repaired.get("referral_reward"),
}
if intent.status in {"cancelled", "expired"}:
raise PaymentCheckoutError(409, f"intent status is {intent.status}")
@@ -2964,6 +3066,7 @@ class PaymentContractCheckoutService:
"duplicate_tx_hash": tx_hash_text,
"payment": repaired.get("payment"),
"subscription": repaired.get("subscription"),
"referral_reward": repaired.get("referral_reward"),
}
raise PaymentCheckoutError(
409, f"intent status is {refreshed.status}, cannot confirm"
@@ -2999,6 +3102,7 @@ class PaymentContractCheckoutService:
plan = self._select_plan(intent.plan_code)
payment_row = {}
subscription_row = {}
referral_reward = {}
try:
payment_row = self._insert_payment_record(
user_id=user_id,
@@ -3015,6 +3119,12 @@ class PaymentContractCheckoutService:
tx_hash=tx_hash_text,
payload=payload,
)
intent.metadata = confirmed_metadata
referral_reward = self._settle_referral_reward_for_intent(
user_id,
intent,
tx_hash_text,
)
except PaymentCheckoutError as exc:
repaired = self._attempt_confirm_repair(
user_id=user_id,
@@ -3025,6 +3135,7 @@ class PaymentContractCheckoutService:
)
payment_row = repaired.get("payment") or payment_row
subscription_row = repaired.get("subscription") or subscription_row
referral_reward = repaired.get("referral_reward") or referral_reward
if not subscription_row:
raise
self._notify_telegram(
@@ -3046,6 +3157,7 @@ class PaymentContractCheckoutService:
"transaction": tx_payload,
"payment": payment_row,
"subscription": subscription_row,
"referral_reward": referral_reward,
"points_redemption": points_result,
"tx": payload,
}