core/engine/src/primitives/workspace.py - main - alexefan136/flowstack

core/engine/src/primitives/workspace.py
920 строк34 KB
Alexander Efanov
upd fix
31 июл 2026, 19:17
31 июл 2026, 19:17d146d86
"""Workspace — пространство для организации пользователей, агентов, flows и ресурсов.

Использует:
- ``src.primitives._time``: timezone-aware UTC
- ``StrEnum`` (Python 3.11+) для статусов/планов/ролей
- Permissions как ``@property`` (вычисляются из роли — нет рассинхрона)
- РФ-биллинг: ``payment_provider`` (cloudpayments / yookassa / stripe)
"""

from __future__ import annotations

from dataclasses import dataclass, field
from datetime import datetime
from enum import StrEnum
from typing import Any
from uuid import uuid4

from src.primitives._time import (
    is_expired as check_expiry,
)
from src.primitives._time import (
    parse_datetime,
    to_iso,
    utc_now,
)

# ============================================================================
# Workspace Enums
# ============================================================================


class WorkspaceStatus(StrEnum):
    """Статус workspace."""

    ACTIVE = "active"
    SUSPENDED = "suspended"
    ARCHIVED = "archived"
    TRIAL = "trial"


class WorkspacePlan(StrEnum):
    """Тарифный план workspace."""

    FREE = "free"
    STARTER = "starter"
    PRO = "pro"
    ENTERPRISE = "enterprise"


class WorkspaceRole(StrEnum):
    """Роли пользователей в workspace."""

    OWNER = "owner"
    ADMIN = "admin"
    MEMBER = "member"
    VIEWER = "viewer"
    GUEST = "guest"


# ============================================================================
# Workspace Member
# ============================================================================


@dataclass
class WorkspaceMember:
    """
    Участник workspace.

    Связывает пользователя с workspace и определяет его роль.
    Permissions вычисляются из роли (``@property``) — всегда актуальны.
    """

    user_id: str
    workspace_id: str
    role: WorkspaceRole = WorkspaceRole.MEMBER

    # Timestamps
    joined_at: datetime = field(default_factory=utc_now)
    invited_at: datetime | None = None
    last_active_at: datetime = field(default_factory=utc_now)

    # Metadata
    email: str | None = None
    display_name: str | None = None
    avatar_url: str | None = None

    # ========================================================================
    # Role checks
    # ========================================================================

    def is_owner(self) -> bool:
        """Проверить, является ли владельцем."""
        return self.role == WorkspaceRole.OWNER

    def is_admin(self) -> bool:
        """Проверить, является ли администратором (или владельцем)."""
        return self.role in (WorkspaceRole.OWNER, WorkspaceRole.ADMIN)

    def is_member(self) -> bool:
        """Проверить, является ли участником (member+)."""
        return self.role in (WorkspaceRole.OWNER, WorkspaceRole.ADMIN, WorkspaceRole.MEMBER)

    def can_edit(self) -> bool:
        """Проверить, может ли редактировать."""
        return self.role in (WorkspaceRole.OWNER, WorkspaceRole.ADMIN, WorkspaceRole.MEMBER)

    def can_view(self) -> bool:
        """Проверить, может ли просматривать."""
        return self.role in (
            WorkspaceRole.OWNER,
            WorkspaceRole.ADMIN,
            WorkspaceRole.MEMBER,
            WorkspaceRole.VIEWER,
            WorkspaceRole.GUEST,
        )

    # ========================================================================
    # Permissions (вычисляются из роли — нет рассинхрона)
    # ========================================================================

    @property
    def can_invite(self) -> bool:
        """Может приглашать участников (owner, admin)."""
        return self.role in (WorkspaceRole.OWNER, WorkspaceRole.ADMIN)

    @property
    def can_manage_members(self) -> bool:
        """Может управлять участниками (owner, admin)."""
        return self.role in (WorkspaceRole.OWNER, WorkspaceRole.ADMIN)

    @property
    def can_manage_billing(self) -> bool:
        """Может управлять биллингом (только owner)."""
        return self.role == WorkspaceRole.OWNER

    @property
    def can_delete_workspace(self) -> bool:
        """Может удалить workspace (только owner)."""
        return self.role == WorkspaceRole.OWNER

    def update_last_active(self) -> None:
        """Обновить время последней активности."""
        self.last_active_at = utc_now()

    def to_dict(self) -> dict[str, Any]:
        """Преобразовать в словарь."""
        return {
            "user_id": self.user_id,
            "workspace_id": self.workspace_id,
            "role": self.role.value,
            "joined_at": to_iso(self.joined_at),
            "invited_at": to_iso(self.invited_at),
            "last_active_at": to_iso(self.last_active_at),
            "email": self.email,
            "display_name": self.display_name,
            "avatar_url": self.avatar_url,
            # Permissions (вычисляемые)
            "can_invite": self.can_invite,
            "can_manage_members": self.can_manage_members,
            "can_manage_billing": self.can_manage_billing,
            "can_delete_workspace": self.can_delete_workspace,
        }

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> WorkspaceMember:
        """Создать из словаря (permissions вычисляются из роли)."""
        return cls(
            user_id=data["user_id"],
            workspace_id=data["workspace_id"],
            role=WorkspaceRole(data.get("role", "member")),
            joined_at=parse_datetime(data.get("joined_at")) or utc_now(),
            invited_at=parse_datetime(data.get("invited_at")),
            last_active_at=parse_datetime(data.get("last_active_at")) or utc_now(),
            email=data.get("email"),
            display_name=data.get("display_name"),
            avatar_url=data.get("avatar_url"),
        )


# ============================================================================
# Workspace Quotas
# ============================================================================


@dataclass
class WorkspaceQuotas:
    """
    Квоты и лимиты для workspace.

    Определяет ограничения в зависимости от тарифного плана.
    """

    # Agents
    max_agents: int = 10
    max_agent_runs_per_day: int = 100

    # Flows
    max_flows: int = 20
    max_flow_runs_per_day: int = 200

    # Tasks
    max_tasks: int = 100
    max_recurring_tasks: int = 10

    # Skills
    max_skills: int = 50
    max_skill_runs_per_day: int = 500

    # MCP Servers
    max_mcp_servers: int = 5
    max_mcp_tool_calls_per_day: int = 1000

    # Storage
    max_storage_mb: int = 1024  # 1 GB
    max_documents: int = 100

    # Members
    max_members: int = 5

    # API
    max_api_requests_per_minute: int = 60
    max_api_requests_per_day: int = 10000

    def can_add_agent(self, current_count: int) -> bool:
        """Проверить, можно ли добавить агента."""
        return current_count < self.max_agents

    def can_add_flow(self, current_count: int) -> bool:
        """Проверить, можно ли добавить flow."""
        return current_count < self.max_flows

    def can_add_task(self, current_count: int) -> bool:
        """Проверить, можно ли добавить задачу."""
        return current_count < self.max_tasks

    def can_add_skill(self, current_count: int) -> bool:
        """Проверить, можно ли добавить скилл."""
        return current_count < self.max_skills

    def can_add_mcp_server(self, current_count: int) -> bool:
        """Проверить, можно ли добавить MCP сервер."""
        return current_count < self.max_mcp_servers

    def can_add_member(self, current_count: int) -> bool:
        """Проверить, можно ли добавить участника."""
        return current_count < self.max_members

    def to_dict(self) -> dict[str, Any]:
        """Преобразовать в словарь."""
        return {
            "max_agents": self.max_agents,
            "max_agent_runs_per_day": self.max_agent_runs_per_day,
            "max_flows": self.max_flows,
            "max_flow_runs_per_day": self.max_flow_runs_per_day,
            "max_tasks": self.max_tasks,
            "max_recurring_tasks": self.max_recurring_tasks,
            "max_skills": self.max_skills,
            "max_skill_runs_per_day": self.max_skill_runs_per_day,
            "max_mcp_servers": self.max_mcp_servers,
            "max_mcp_tool_calls_per_day": self.max_mcp_tool_calls_per_day,
            "max_storage_mb": self.max_storage_mb,
            "max_documents": self.max_documents,
            "max_members": self.max_members,
            "max_api_requests_per_minute": self.max_api_requests_per_minute,
            "max_api_requests_per_day": self.max_api_requests_per_day,
        }

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> WorkspaceQuotas:
        """Создать из словаря."""
        return cls(
            max_agents=data.get("max_agents", 10),
            max_agent_runs_per_day=data.get("max_agent_runs_per_day", 100),
            max_flows=data.get("max_flows", 20),
            max_flow_runs_per_day=data.get("max_flow_runs_per_day", 200),
            max_tasks=data.get("max_tasks", 100),
            max_recurring_tasks=data.get("max_recurring_tasks", 10),
            max_skills=data.get("max_skills", 50),
            max_skill_runs_per_day=data.get("max_skill_runs_per_day", 500),
            max_mcp_servers=data.get("max_mcp_servers", 5),
            max_mcp_tool_calls_per_day=data.get("max_mcp_tool_calls_per_day", 1000),
            max_storage_mb=data.get("max_storage_mb", 1024),
            max_documents=data.get("max_documents", 100),
            max_members=data.get("max_members", 5),
            max_api_requests_per_minute=data.get("max_api_requests_per_minute", 60),
            max_api_requests_per_day=data.get("max_api_requests_per_day", 10000),
        )

    @classmethod
    def for_plan(cls, plan: WorkspacePlan) -> WorkspaceQuotas:
        """Создать квоты для тарифного плана."""
        if plan == WorkspacePlan.FREE:
            return cls(
                max_agents=5,
                max_agent_runs_per_day=50,
                max_flows=10,
                max_flow_runs_per_day=100,
                max_tasks=50,
                max_recurring_tasks=5,
                max_skills=20,
                max_skill_runs_per_day=200,
                max_mcp_servers=2,
                max_mcp_tool_calls_per_day=500,
                max_storage_mb=512,
                max_documents=50,
                max_members=3,
                max_api_requests_per_minute=30,
                max_api_requests_per_day=5000,
            )
        if plan == WorkspacePlan.STARTER:
            return cls(
                max_agents=20,
                max_agent_runs_per_day=500,
                max_flows=50,
                max_flow_runs_per_day=1000,
                max_tasks=500,
                max_recurring_tasks=50,
                max_skills=100,
                max_skill_runs_per_day=2000,
                max_mcp_servers=10,
                max_mcp_tool_calls_per_day=5000,
                max_storage_mb=5120,  # 5 GB
                max_documents=500,
                max_members=10,
                max_api_requests_per_minute=120,
                max_api_requests_per_day=50000,
            )
        if plan == WorkspacePlan.PRO:
            return cls(
                max_agents=100,
                max_agent_runs_per_day=5000,
                max_flows=200,
                max_flow_runs_per_day=10000,
                max_tasks=5000,
                max_recurring_tasks=500,
                max_skills=500,
                max_skill_runs_per_day=20000,
                max_mcp_servers=50,
                max_mcp_tool_calls_per_day=50000,
                max_storage_mb=51200,  # 50 GB
                max_documents=5000,
                max_members=50,
                max_api_requests_per_minute=600,
                max_api_requests_per_day=500000,
            )
        # ENTERPRISE
        return cls(
            max_agents=10000,
            max_agent_runs_per_day=100000,
            max_flows=10000,
            max_flow_runs_per_day=1000000,
            max_tasks=100000,
            max_recurring_tasks=10000,
            max_skills=10000,
            max_skill_runs_per_day=1000000,
            max_mcp_servers=1000,
            max_mcp_tool_calls_per_day=10000000,
            max_storage_mb=1048576,  # 1 TB
            max_documents=100000,
            max_members=1000,
            max_api_requests_per_minute=6000,
            max_api_requests_per_day=10000000,
        )


# ============================================================================
# Workspace Usage
# ============================================================================


@dataclass
class WorkspaceUsage:
    """
    Использование ресурсов workspace.

    Отслеживает текущее использование для проверки квот.
    """

    # Counts
    agent_count: int = 0
    flow_count: int = 0
    task_count: int = 0
    skill_count: int = 0
    mcp_server_count: int = 0
    member_count: int = 0
    document_count: int = 0

    # Daily usage
    agent_runs_today: int = 0
    flow_runs_today: int = 0
    skill_runs_today: int = 0
    mcp_tool_calls_today: int = 0
    api_requests_today: int = 0

    # Storage
    storage_used_mb: float = 0.0

    # Timestamps
    last_reset_at: datetime = field(default_factory=utc_now)

    def increment_agent_runs(self) -> None:
        """Инкрементировать счётчик запусков агентов."""
        self.agent_runs_today += 1

    def increment_flow_runs(self) -> None:
        """Инкрементировать счётчик запусков flows."""
        self.flow_runs_today += 1

    def increment_skill_runs(self) -> None:
        """Инкрементировать счётчик запусков скиллов."""
        self.skill_runs_today += 1

    def increment_mcp_tool_calls(self) -> None:
        """Инкрементировать счётчик вызовов MCP tools."""
        self.mcp_tool_calls_today += 1

    def increment_api_requests(self) -> None:
        """Инкрементировать счётчик API запросов."""
        self.api_requests_today += 1

    def reset_daily_counters(self) -> None:
        """Сбросить дневные счётчики."""
        self.agent_runs_today = 0
        self.flow_runs_today = 0
        self.skill_runs_today = 0
        self.mcp_tool_calls_today = 0
        self.api_requests_today = 0
        self.last_reset_at = utc_now()

    def check_limits(self, quotas: WorkspaceQuotas) -> list[str]:
        """
        Проверить, не превышены ли лимиты.

        Returns:
            Список нарушений (пустой если всё ок).
        """
        violations: list[str] = []

        if self.agent_count >= quotas.max_agents:
            violations.append(f"Agent limit exceeded: {self.agent_count}/{quotas.max_agents}")
        if self.flow_count >= quotas.max_flows:
            violations.append(f"Flow limit exceeded: {self.flow_count}/{quotas.max_flows}")
        if self.task_count >= quotas.max_tasks:
            violations.append(f"Task limit exceeded: {self.task_count}/{quotas.max_tasks}")
        if self.skill_count >= quotas.max_skills:
            violations.append(f"Skill limit exceeded: {self.skill_count}/{quotas.max_skills}")
        if self.mcp_server_count >= quotas.max_mcp_servers:
            violations.append(
                f"MCP server limit exceeded: {self.mcp_server_count}/{quotas.max_mcp_servers}"
            )
        if self.member_count >= quotas.max_members:
            violations.append(f"Member limit exceeded: {self.member_count}/{quotas.max_members}")
        if self.document_count >= quotas.max_documents:
            violations.append(
                f"Document limit exceeded: {self.document_count}/{quotas.max_documents}"
            )
        if self.agent_runs_today >= quotas.max_agent_runs_per_day:
            violations.append(
                f"Daily agent runs limit exceeded: "
                f"{self.agent_runs_today}/{quotas.max_agent_runs_per_day}"
            )
        if self.flow_runs_today >= quotas.max_flow_runs_per_day:
            violations.append(
                f"Daily flow runs limit exceeded: "
                f"{self.flow_runs_today}/{quotas.max_flow_runs_per_day}"
            )
        if self.skill_runs_today >= quotas.max_skill_runs_per_day:
            violations.append(
                f"Daily skill runs limit exceeded: "
                f"{self.skill_runs_today}/{quotas.max_skill_runs_per_day}"
            )
        if self.mcp_tool_calls_today >= quotas.max_mcp_tool_calls_per_day:
            violations.append(
                f"Daily MCP tool calls limit exceeded: "
                f"{self.mcp_tool_calls_today}/{quotas.max_mcp_tool_calls_per_day}"
            )
        if self.api_requests_today >= quotas.max_api_requests_per_day:
            violations.append(
                f"Daily API requests limit exceeded: "
                f"{self.api_requests_today}/{quotas.max_api_requests_per_day}"
            )
        if self.storage_used_mb >= quotas.max_storage_mb:
            violations.append(
                f"Storage limit exceeded: {self.storage_used_mb:.1f}/{quotas.max_storage_mb} MB"
            )

        return violations

    def to_dict(self) -> dict[str, Any]:
        """Преобразовать в словарь."""
        return {
            "agent_count": self.agent_count,
            "flow_count": self.flow_count,
            "task_count": self.task_count,
            "skill_count": self.skill_count,
            "mcp_server_count": self.mcp_server_count,
            "member_count": self.member_count,
            "document_count": self.document_count,
            "agent_runs_today": self.agent_runs_today,
            "flow_runs_today": self.flow_runs_today,
            "skill_runs_today": self.skill_runs_today,
            "mcp_tool_calls_today": self.mcp_tool_calls_today,
            "api_requests_today": self.api_requests_today,
            "storage_used_mb": self.storage_used_mb,
            "last_reset_at": to_iso(self.last_reset_at),
        }

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> WorkspaceUsage:
        """Создать из словаря."""
        return cls(
            agent_count=data.get("agent_count", 0),
            flow_count=data.get("flow_count", 0),
            task_count=data.get("task_count", 0),
            skill_count=data.get("skill_count", 0),
            mcp_server_count=data.get("mcp_server_count", 0),
            member_count=data.get("member_count", 0),
            document_count=data.get("document_count", 0),
            agent_runs_today=data.get("agent_runs_today", 0),
            flow_runs_today=data.get("flow_runs_today", 0),
            skill_runs_today=data.get("skill_runs_today", 0),
            mcp_tool_calls_today=data.get("mcp_tool_calls_today", 0),
            api_requests_today=data.get("api_requests_today", 0),
            storage_used_mb=data.get("storage_used_mb", 0.0),
            last_reset_at=parse_datetime(data.get("last_reset_at")) or utc_now(),
        )


# ============================================================================
# Workspace Settings
# ============================================================================


@dataclass
class WorkspaceSettings:
    """
    Настройки workspace.

    Содержит конфигурацию и предпочтения workspace.
    """

    # General
    name: str = "Default Workspace"
    description: str = ""
    logo_url: str | None = None

    # Defaults
    default_model: str = "deepseek-ai/DeepSeek-V4-Pro"
    default_language: str = "ru"
    timezone: str = "UTC"

    # Features
    rag_enabled: bool = True
    agents_enabled: bool = True
    flows_enabled: bool = True
    tasks_enabled: bool = True
    skills_enabled: bool = True
    mcp_servers_enabled: bool = True

    # Security
    require_2fa: bool = False
    session_timeout_minutes: int = 60
    password_min_length: int = 8

    # Notifications
    email_notifications_enabled: bool = True
    slack_notifications_enabled: bool = False
    webhook_url: str | None = None

    # Metadata
    custom_metadata: dict[str, Any] = field(default_factory=dict)

    def get_setting(self, key: str, default: Any = None) -> Any:
        """Получить настройку из custom_metadata."""
        return self.custom_metadata.get(key, default)

    def set_setting(self, key: str, value: Any) -> None:
        """Установить настройку в custom_metadata."""
        self.custom_metadata[key] = value

    def to_dict(self) -> dict[str, Any]:
        """Преобразовать в словарь."""
        return {
            "name": self.name,
            "description": self.description,
            "logo_url": self.logo_url,
            "default_model": self.default_model,
            "default_language": self.default_language,
            "timezone": self.timezone,
            "rag_enabled": self.rag_enabled,
            "agents_enabled": self.agents_enabled,
            "flows_enabled": self.flows_enabled,
            "tasks_enabled": self.tasks_enabled,
            "skills_enabled": self.skills_enabled,
            "mcp_servers_enabled": self.mcp_servers_enabled,
            "require_2fa": self.require_2fa,
            "session_timeout_minutes": self.session_timeout_minutes,
            "password_min_length": self.password_min_length,
            "email_notifications_enabled": self.email_notifications_enabled,
            "slack_notifications_enabled": self.slack_notifications_enabled,
            "webhook_url": self.webhook_url,
            "custom_metadata": self.custom_metadata,
        }

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> WorkspaceSettings:
        """Создать из словаря."""
        return cls(
            name=data.get("name", "Default Workspace"),
            description=data.get("description", ""),
            logo_url=data.get("logo_url"),
            default_model=data.get("default_model", "deepseek-ai/DeepSeek-V4-Pro"),
            default_language=data.get("default_language", "ru"),
            timezone=data.get("timezone", "UTC"),
            rag_enabled=data.get("rag_enabled", True),
            agents_enabled=data.get("agents_enabled", True),
            flows_enabled=data.get("flows_enabled", True),
            tasks_enabled=data.get("tasks_enabled", True),
            skills_enabled=data.get("skills_enabled", True),
            mcp_servers_enabled=data.get("mcp_servers_enabled", True),
            require_2fa=data.get("require_2fa", False),
            session_timeout_minutes=data.get("session_timeout_minutes", 60),
            password_min_length=data.get("password_min_length", 8),
            email_notifications_enabled=data.get("email_notifications_enabled", True),
            slack_notifications_enabled=data.get("slack_notifications_enabled", False),
            webhook_url=data.get("webhook_url"),
            custom_metadata=data.get("custom_metadata", {}),
        )


# ============================================================================
# Workspace
# ============================================================================


@dataclass
class Workspace:
    """
    Workspace — пространство для организации ресурсов.

    Содержит: информацию, участников и роли, квоты и использование, настройки.
    """

    id: str = field(default_factory=lambda: str(uuid4()))
    name: str = "Default Workspace"
    description: str = ""

    # Owner
    owner_id: str = ""

    # Status
    status: WorkspaceStatus = WorkspaceStatus.ACTIVE
    plan: WorkspacePlan = WorkspacePlan.FREE

    # Members
    members: list[WorkspaceMember] = field(default_factory=list)

    # Quotas and usage
    quotas: WorkspaceQuotas = field(default_factory=WorkspaceQuotas)
    usage: WorkspaceUsage = field(default_factory=WorkspaceUsage)

    # Settings
    settings: WorkspaceSettings = field(default_factory=WorkspaceSettings)

    # Billing (РФ + международный)
    billing_email: str | None = None
    payment_provider: str | None = None  # cloudpayments, yookassa, stripe
    payment_customer_id: str | None = None  # ID клиента в платёжной системе
    trial_ends_at: datetime | None = None

    # Timestamps
    created_at: datetime = field(default_factory=utc_now)
    updated_at: datetime = field(default_factory=utc_now)

    # ========================================================================
    # Members
    # ========================================================================

    def add_member(self, member: WorkspaceMember) -> bool:
        """
        Добавить участника.

        Проверяет:
        - Отсутствие дубликата
        - Квоту на количество участников

        Returns:
            True если добавлен, False если дубликат или квота исчерпана.
        """
        if any(m.user_id == member.user_id for m in self.members):
            return False
        if not self.quotas.can_add_member(len(self.members)):
            return False

        self.members.append(member)
        self.usage.member_count = len(self.members)
        self.updated_at = utc_now()
        return True

    def remove_member(self, user_id: str) -> bool:
        """Удалить участника (нельзя удалить владельца)."""
        member = self.get_member(user_id)
        if not member:
            return False
        if member.is_owner():
            return False

        self.members.remove(member)
        self.usage.member_count = len(self.members)
        self.updated_at = utc_now()
        return True

    def get_member(self, user_id: str) -> WorkspaceMember | None:
        """Получить участника по user_id."""
        return next((m for m in self.members if m.user_id == user_id), None)

    def get_owner(self) -> WorkspaceMember | None:
        """Получить владельца."""
        return next((m for m in self.members if m.is_owner()), None)

    def get_admins(self) -> list[WorkspaceMember]:
        """Получить список администраторов."""
        return [m for m in self.members if m.is_admin()]

    def transfer_ownership(self, new_owner_id: str) -> bool:
        """
        Передать владение workspace другому участнику.

        Текущий владелец становится админом, новый — владельцем.

        Returns:
            True если передача успешна, False если новый владелец не участник.
        """
        new_owner = self.get_member(new_owner_id)
        if not new_owner:
            return False

        current_owner = self.get_owner()
        if current_owner:
            current_owner.role = WorkspaceRole.ADMIN

        new_owner.role = WorkspaceRole.OWNER
        self.owner_id = new_owner_id
        self.updated_at = utc_now()
        return True

    # ========================================================================
    # Permission checks
    # ========================================================================

    def is_member(self, user_id: str) -> bool:
        """Проверить, является ли пользователь участником."""
        return self.get_member(user_id) is not None

    def is_owner(self, user_id: str) -> bool:
        """Проверить, является ли пользователь владельцем."""
        member = self.get_member(user_id)
        return member is not None and member.is_owner()

    def is_admin(self, user_id: str) -> bool:
        """Проверить, является ли пользователь администратором."""
        member = self.get_member(user_id)
        return member is not None and member.is_admin()

    def can_user_edit(self, user_id: str) -> bool:
        """Проверить, может ли пользователь редактировать."""
        member = self.get_member(user_id)
        return member is not None and member.can_edit()

    def can_user_view(self, user_id: str) -> bool:
        """Проверить, может ли пользователь просматривать."""
        member = self.get_member(user_id)
        return member is not None and member.can_view()

    # ========================================================================
    # Quotas & Status
    # ========================================================================

    def check_quotas(self) -> list[str]:
        """Проверить, не превышены ли квоты."""
        return self.usage.check_limits(self.quotas)

    def is_trial_expired(self) -> bool:
        """Проверить, истёк ли пробный период."""
        return check_expiry(self.trial_ends_at)

    def is_active(self) -> bool:
        """Проверить, активен ли workspace."""
        return self.status == WorkspaceStatus.ACTIVE

    def suspend(self) -> None:
        """Приостановить workspace."""
        self.status = WorkspaceStatus.SUSPENDED
        self.updated_at = utc_now()

    def activate(self) -> None:
        """Активировать workspace."""
        self.status = WorkspaceStatus.ACTIVE
        self.updated_at = utc_now()

    def archive(self) -> None:
        """Архивировать workspace."""
        self.status = WorkspaceStatus.ARCHIVED
        self.updated_at = utc_now()

    # ========================================================================
    # Serialization
    # ========================================================================

    def to_dict(self) -> dict[str, Any]:
        """Преобразовать в словарь."""
        return {
            "id": self.id,
            "name": self.name,
            "description": self.description,
            "owner_id": self.owner_id,
            "status": self.status.value,
            "plan": self.plan.value,
            "members": [m.to_dict() for m in self.members],
            "quotas": self.quotas.to_dict(),
            "usage": self.usage.to_dict(),
            "settings": self.settings.to_dict(),
            "billing_email": self.billing_email,
            "payment_provider": self.payment_provider,
            "payment_customer_id": self.payment_customer_id,
            "trial_ends_at": to_iso(self.trial_ends_at),
            "created_at": to_iso(self.created_at),
            "updated_at": to_iso(self.updated_at),
        }

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> Workspace:
        """Создать из словаря."""
        return cls(
            id=data.get("id", str(uuid4())),
            name=data.get("name", "Default Workspace"),
            description=data.get("description", ""),
            owner_id=data.get("owner_id", ""),
            status=WorkspaceStatus(data.get("status", "active")),
            plan=WorkspacePlan(data.get("plan", "free")),
            members=[WorkspaceMember.from_dict(m) for m in data.get("members", [])],
            quotas=WorkspaceQuotas.from_dict(data.get("quotas", {})),
            usage=WorkspaceUsage.from_dict(data.get("usage", {})),
            settings=WorkspaceSettings.from_dict(data.get("settings", {})),
            billing_email=data.get("billing_email"),
            payment_provider=data.get("payment_provider"),
            payment_customer_id=data.get("payment_customer_id"),
            trial_ends_at=parse_datetime(data.get("trial_ends_at")),
            created_at=parse_datetime(data.get("created_at")) or utc_now(),
            updated_at=parse_datetime(data.get("updated_at")) or utc_now(),
        )


# ============================================================================
# Helper Functions
# ============================================================================


def create_workspace(
    name: str,
    owner_id: str,
    plan: WorkspacePlan = WorkspacePlan.FREE,
    **kwargs: Any,
) -> Workspace:
    """
    Создать новый workspace.

    Автоматически добавляет владельца как участника с ролью OWNER
    (permissions вычисляются из роли).
    """
    workspace = Workspace(
        name=name,
        owner_id=owner_id,
        plan=plan,
        quotas=WorkspaceQuotas.for_plan(plan),
        **kwargs,
    )

    owner_member = WorkspaceMember(
        user_id=owner_id,
        workspace_id=workspace.id,
        role=WorkspaceRole.OWNER,
    )
    workspace.add_member(owner_member)

    return workspace


def create_workspace_member(
    user_id: str,
    workspace_id: str,
    role: WorkspaceRole = WorkspaceRole.MEMBER,
    **kwargs: Any,
) -> WorkspaceMember:
    """Создать участника workspace (permissions вычисляются из роли)."""
    return WorkspaceMember(
        user_id=user_id,
        workspace_id=workspace_id,
        role=role,
        **kwargs,
    )


# ============================================================================
# Exports
# ============================================================================

__all__ = [
    "Workspace",
    "WorkspaceMember",
    "WorkspacePlan",
    "WorkspaceQuotas",
    "WorkspaceRole",
    "WorkspaceSettings",
    "WorkspaceStatus",
    "WorkspaceUsage",
    "create_workspace",
    "create_workspace_member",
]