224 lines
6.9 KiB
Python
224 lines
6.9 KiB
Python
"""
|
|
Subscription 管理 Use Case
|
|
"""
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import Optional
|
|
|
|
from packages.domain.entities import WorkspaceMemberRole
|
|
|
|
|
|
class UpgradeSubscriptionRequest:
|
|
"""升级订阅请求"""
|
|
|
|
def __init__(
|
|
self,
|
|
workspace_id: str,
|
|
requester_user_id: str,
|
|
new_plan: str,
|
|
):
|
|
self.workspace_id = workspace_id
|
|
self.requester_user_id = requester_user_id
|
|
self.new_plan = new_plan
|
|
|
|
|
|
class UpgradeSubscriptionResponse:
|
|
"""升级订阅响应"""
|
|
|
|
def __init__(
|
|
self,
|
|
workspace_id: str,
|
|
old_plan: str,
|
|
new_plan: str,
|
|
max_projects: int,
|
|
max_storage_gb: int,
|
|
):
|
|
self.workspace_id = workspace_id
|
|
self.old_plan = old_plan
|
|
self.new_plan = new_plan
|
|
self.max_projects = max_projects
|
|
self.max_storage_gb = max_storage_gb
|
|
|
|
|
|
class UpgradeSubscriptionUseCase:
|
|
"""升级订阅用例"""
|
|
|
|
# 订阅计划配额
|
|
PLAN_QUOTAS = {
|
|
"free": {"max_projects": 3, "max_storage_gb": 10, "price": 0},
|
|
"pro": {"max_projects": 999999, "max_storage_gb": 100, "price": 99},
|
|
"enterprise": {"max_projects": 999999, "max_storage_gb": 1000, "price": 999},
|
|
}
|
|
|
|
# 计划等级
|
|
PLAN_LEVELS = {
|
|
"free": 0,
|
|
"pro": 1,
|
|
"enterprise": 2,
|
|
}
|
|
|
|
def __init__(
|
|
self,
|
|
workspace_repository,
|
|
workspace_member_repository,
|
|
):
|
|
self.workspace_repository = workspace_repository
|
|
self.workspace_member_repository = workspace_member_repository
|
|
|
|
def execute(
|
|
self, request: UpgradeSubscriptionRequest
|
|
) -> tuple[Optional[UpgradeSubscriptionResponse], Optional[str]]:
|
|
"""
|
|
执行升级订阅
|
|
|
|
Args:
|
|
request: 升级请求
|
|
|
|
Returns:
|
|
(响应, 错误信息)
|
|
"""
|
|
try:
|
|
# 1. 验证输入
|
|
if not request.workspace_id:
|
|
return None, "Workspace ID is required"
|
|
|
|
if not request.requester_user_id:
|
|
return None, "Requester user ID is required"
|
|
|
|
if not request.new_plan:
|
|
return None, "New plan is required"
|
|
|
|
# 2. 验证新计划有效
|
|
if request.new_plan not in self.PLAN_QUOTAS:
|
|
return None, f"Invalid plan: {request.new_plan}"
|
|
|
|
# 3. 验证工作空间存在
|
|
workspace = self.workspace_repository.find_by_id(request.workspace_id)
|
|
if not workspace:
|
|
return None, "Workspace not found"
|
|
|
|
# 4. 验证权限（只有 Owner 可以管理订阅）
|
|
member = self.workspace_member_repository.find_by_workspace_and_user(
|
|
request.workspace_id,
|
|
request.requester_user_id,
|
|
)
|
|
if not member:
|
|
return None, "You are not a member of this workspace"
|
|
|
|
if member.role != WorkspaceMemberRole.OWNER:
|
|
return None, "Only workspace owner can manage subscription"
|
|
|
|
# 5. 检查是否是升级（不能降级到免费计划，需要用取消订阅）
|
|
current_level = self.PLAN_LEVELS.get(workspace.subscription_plan, 0)
|
|
new_level = self.PLAN_LEVELS.get(request.new_plan, 0)
|
|
|
|
if new_level < current_level:
|
|
return (
|
|
None,
|
|
"Cannot downgrade plan. Use cancel subscription to return to free plan.",
|
|
)
|
|
|
|
if new_level == current_level:
|
|
return None, f"Workspace is already on {request.new_plan} plan"
|
|
|
|
# 6. 更新订阅
|
|
old_plan = workspace.subscription_plan
|
|
quota = self.PLAN_QUOTAS[request.new_plan]
|
|
|
|
workspace.subscription_plan = request.new_plan
|
|
workspace.subscription_status = "active"
|
|
workspace.max_projects = quota["max_projects"]
|
|
workspace.max_storage_gb = quota["max_storage_gb"]
|
|
|
|
# 设置过期时间（假设按月订阅）
|
|
workspace.subscription_expires_at = datetime.now(timezone.utc) + timedelta(days=30)
|
|
|
|
self.workspace_repository.save(workspace)
|
|
|
|
# 7. 返回响应
|
|
return (
|
|
UpgradeSubscriptionResponse(
|
|
workspace_id=workspace.id,
|
|
old_plan=old_plan,
|
|
new_plan=workspace.subscription_plan,
|
|
max_projects=workspace.max_projects,
|
|
max_storage_gb=workspace.max_storage_gb,
|
|
),
|
|
None,
|
|
)
|
|
|
|
except Exception as e:
|
|
return None, f"Failed to upgrade subscription: {str(e)}"
|
|
|
|
|
|
class CancelSubscriptionRequest:
|
|
"""取消订阅请求"""
|
|
|
|
def __init__(self, workspace_id: str, requester_user_id: str):
|
|
self.workspace_id = workspace_id
|
|
self.requester_user_id = requester_user_id
|
|
|
|
|
|
class CancelSubscriptionUseCase:
|
|
"""取消订阅用例"""
|
|
|
|
def __init__(
|
|
self,
|
|
workspace_repository,
|
|
workspace_member_repository,
|
|
):
|
|
self.workspace_repository = workspace_repository
|
|
self.workspace_member_repository = workspace_member_repository
|
|
|
|
def execute(self, request: CancelSubscriptionRequest) -> tuple[bool, Optional[str]]:
|
|
"""
|
|
执行取消订阅
|
|
|
|
Args:
|
|
request: 取消请求
|
|
|
|
Returns:
|
|
(是否成功, 错误信息)
|
|
"""
|
|
try:
|
|
# 1. 验证输入
|
|
if not request.workspace_id:
|
|
return False, "Workspace ID is required"
|
|
|
|
if not request.requester_user_id:
|
|
return False, "Requester user ID is required"
|
|
|
|
# 2. 验证工作空间存在
|
|
workspace = self.workspace_repository.find_by_id(request.workspace_id)
|
|
if not workspace:
|
|
return False, "Workspace not found"
|
|
|
|
# 3. 验证权限（只有 Owner 可以管理订阅）
|
|
member = self.workspace_member_repository.find_by_workspace_and_user(
|
|
request.workspace_id,
|
|
request.requester_user_id,
|
|
)
|
|
if not member:
|
|
return False, "You are not a member of this workspace"
|
|
|
|
if member.role != WorkspaceMemberRole.OWNER:
|
|
return False, "Only workspace owner can manage subscription"
|
|
|
|
# 4. 检查当前计划
|
|
if workspace.subscription_plan == "free":
|
|
return False, "Workspace is already on free plan"
|
|
|
|
# 5. 降级到 free 计划
|
|
workspace.subscription_plan = "free"
|
|
workspace.subscription_status = "active"
|
|
workspace.subscription_expires_at = None
|
|
workspace.max_projects = 3
|
|
workspace.max_storage_gb = 10
|
|
|
|
self.workspace_repository.save(workspace)
|
|
|
|
return True, None
|
|
|
|
except Exception as e:
|
|
return False, f"Failed to cancel subscription: {str(e)}"
|