xiaoxia-saas/packages/adapters/in_memory/project_repository.py

from __future__ import annotations

from packages.domain import Project


class InMemoryProjectRepository:
    def __init__(self):
        self._items: dict[str, Project] = {}  # project_id -> Project

    def save(self, project: Project) -> Project:
        self._items[project.id] = project
        return project

    def find_by_id(self, project_id: str) -> Project | None:
        return self._items.get(project_id)

    def find_by_owner_user_id(self, owner_user_id: str) -> list[Project]:
        return [p for p in self._items.values() if p.owner_user_id == owner_user_id]

    def find_accessible_projects(self, user_id: str) -> list[Project]:
        """查找用户可访问的所有项目（自己拥有的 + 被共享的）"""
        return [p for p in self._items.values() if p.can_access(user_id)]

    def count_by_owner(self, owner_user_id: str) -> int:
        return len([p for p in self._items.values() if p.owner_user_id == owner_user_id])

    def delete(self, project_id: str) -> bool:
        if project_id in self._items:
            del self._items[project_id]
            return True
        return False

    def find_default_by_owner(self, owner_user_id: str) -> Project | None:
        """查找用户的默认项目（Issue #1775 幂等接口，内存实现）。"""
        for p in self._items.values():
            if p.owner_user_id == owner_user_id and getattr(p, "is_default", False):
                return p
        return None

    def get_or_create_default_project(
        self,
        owner_user_id: str,
        *,
        name: str = "默认项目",
        description: str = "小程序自动创建的默认项目",
    ) -> Project:
        """幂等获取/创建默认项目（内存实现，模拟 DB 部分唯一索引语义）。"""
        existing = self.find_default_by_owner(owner_user_id)
        if existing is not None:
            return existing
        project = Project.create(
            owner_user_id=owner_user_id,
            name=name,
            description=description,
            is_default=True,
        )
        self._items[project.id] = project
        return project