docs/v1.0/server/changelogs/2026-01-27-service-docs-uuid-cleanup.md

服务设计文档 UUID 定义清理

变更日期：2026-01-27
变更类型：文档更新
影响范围：docs/requirements/backend/04-services/ 下所有服务文档

---

变更概述

批量清理所有服务设计文档中过时的数据库定义，移除物理外键约束和 UUID v7 数据库默认值，与项目架构规范保持一致。

---

变更内容

1. 移除物理外键约束

涉及文档：15 个服务文档

修改内容：

• 移除所有 REFERENCES ... ON DELETE CASCADE
• 移除所有 REFERENCES ... ON DELETE SET NULL
• 移除所有 REFERENCES ... ON DELETE RESTRICT
• 移除所有 REFERENCES users(user_id)
• 移除所有 REFERENCES users(id)

示例：

-- ❌ 修改前
created_by UUID NOT NULL REFERENCES users(user_id),
user_id UUID NOT NULL REFERENCES users(user_id) ON DELETE CASCADE,
category_id UUID NOT NULL REFERENCES screenplay_tag_categories(category_id) ON DELETE RESTRICT,

-- ✅ 修改后
created_by UUID NOT NULL,
user_id UUID NOT NULL,
category_id UUID NOT NULL,

2. 移除 UUID v7 数据库默认值

涉及文档：所有包含表定义的文档

修改内容：

• 移除所有 DEFAULT gen_uuid_v7()
• 移除所有 server_default=text("gen_uuid_v7()")

示例：

-- ❌ 修改前
user_id UUID PRIMARY KEY DEFAULT gen_uuid_v7(),

-- ✅ 修改后
user_id UUID PRIMARY KEY,

# ❌ 修改前
user_id: UUID = Field(
    primary_key=True,
    server_default=text("gen_uuid_v7()")
)

# ✅ 修改后
user_id: UUID = Field(
    primary_key=True,
    default_factory=generate_uuid
)

3. 更新约束说明表格

文档：docs/requirements/backend/04-services/user/user-service.md

修改内容：

• 移除 REFERENCES 外键约束示例
• 添加项目架构约束说明

修改前：

约束类型	说明	示例
REFERENCES	外键约束	avatar_id REFERENCES attachments(attachment_id)

修改后：

移除该行，并添加说明：

注意：本项目禁止使用数据库物理外键约束（REFERENCES），所有引用完整性在应用层 Service 中验证。

---

涉及文件清单

已修改的服务文档（15 个）

1. docs/requirements/backend/04-services/user/user-service.md
2. docs/requirements/backend/04-services/project/screenplay-tag-service.md
3. docs/requirements/backend/04-services/project/project-resource-service.md
4. docs/requirements/backend/04-services/project/folder-service.md
5. docs/requirements/backend/04-services/project/project-service.md
6. docs/requirements/backend/04-services/project/screenplay-service.md
7. docs/requirements/backend/04-services/resource/attachment-service.md
8. docs/requirements/backend/04-services/resource/project-resource-service.md
9. docs/requirements/backend/04-services/ai/video-service.md
10. 其他包含过时定义的服务文档

---

验证结果

验证命令

# 验证不再包含物理外键约束
grep -r "REFERENCES.*(" docs/requirements/backend/04-services --include="*.md" | grep -v "references/" | grep -v "参考文档"

# 验证不再包含 UUID v7 数据库默认值
grep -r "DEFAULT gen_uuid_v7()" docs/requirements/backend/04-services --include="*.md"
grep -r "server_default.*gen_uuid_v7" docs/requirements/backend/04-services --include="*.md"

验证结果

✅ 所有 SQL 定义中不再包含物理外键约束
✅ 所有 SQL 定义中不再包含 UUID v7 数据库默认值
✅ 所有 Python Model 定义中不再包含 server_default

---

架构规范说明

项目数据库设计约束

根据 .claude/skills/jointo-tech-stack/references/database.md：

1. 禁止物理外键约束：

   • 数据库层禁止创建 FOREIGN KEY 约束
   • 所有引用完整性在应用层 Service/Repository 中验证
   • 所有逻辑关联字段必须手动创建索引

2. 禁止数据库默认值生成 UUID：

   • 禁止使用 DEFAULT gen_uuid_v7()
   • 禁止使用 server_default=text("gen_uuid_v7()")
   • 必须在应用层通过 default_factory=generate_uuid 生成

3. 原因：

   • 为分库分表做准备
   • 提升数据库性能
   • 增强系统扩展性
   • 统一 UUID 生成逻辑

---

应用层关联关系处理

移除物理外键约束后，需要在应用层（Service/Repository）保证引用完整性和处理级联关系。

三层保证机制

1. Repository 层 - 基础验证

提供通用的引用检查方法：

# repositories/base_repository.py
class BaseRepository(Generic[T]):
    async def exists(self, id: UUID) -> bool:
        """检查记录是否存在（排除软删除）"""
        query = select(self.model.id).where(
            self.model.id == id,
            self.model.deleted_at.is_(None)
        ).limit(1)
        result = await self.db.execute(query)
        return result.scalar_one_or_none() is not None
    
    async def batch_exists(self, ids: list[UUID]) -> dict[UUID, bool]:
        """批量检查记录是否存在（性能优化）"""
        query = select(self.model.id).where(
            self.model.id.in_(ids),
            self.model.deleted_at.is_(None)
        )
        result = await self.db.execute(query)
        existing_ids = {row[0] for row in result.all()}
        return {id: id in existing_ids for id in ids}

2. Service 层 - 业务验证

在所有涉及关联关系的操作中验证引用完整性：

创建操作 - 验证所有外键引用

# services/project_service.py
async def create_project(self, user_id: UUID, data: ProjectCreate) -> Project:
    """创建项目 - 验证所有引用"""
    
    # 1. 验证所有者是否存在
    if not await self.user_repo.exists(data.owner_id):
        raise NotFoundError("所有者用户不存在")
    
    # 2. 验证文件夹引用（如果提供）
    if data.folder_id:
        folder = await self.folder_repo.get_by_id(data.folder_id)
        if not folder:
            raise NotFoundError("文件夹不存在")
        
        # 验证用户对文件夹的权限
        has_permission = await self.folder_repo.check_user_permission(
            user_id, data.folder_id, MemberRole.EDITOR
        )
        if not has_permission:
            raise PermissionDeniedError("无权限在该文件夹下创建项目")
    
    # 3. 使用事务创建项目和关联数据
    async with self.db.begin():
        project = await self.project_repo.create(data)
        
        # 自动添加创建者为所有者成员
        await self.project_member_repo.add_member(
            project_id=project.id,
            user_id=user_id,
            role=MemberRole.OWNER
        )
    
    return project

更新操作 - 验证新的外键引用

async def update_project(
    self,
    user_id: UUID,
    project_id: UUID,
    data: ProjectUpdate
) -> Project:
    """更新项目 - 验证新的引用"""
    
    # 1. 验证项目是否存在
    project = await self.project_repo.get_by_id(project_id)
    if not project:
        raise NotFoundError("项目不存在")
    
    # 2. 如果更新文件夹，验证新文件夹
    if data.folder_id is not None and data.folder_id != project.folder_id:
        if not await self.folder_repo.exists(data.folder_id):
            raise NotFoundError("目标文件夹不存在")
        
        # 验证用户对新文件夹的权限
        has_permission = await self.folder_repo.check_user_permission(
            user_id, data.folder_id, MemberRole.EDITOR
        )
        if not has_permission:
            raise PermissionDeniedError("无权限移动到该文件夹")
    
    # 3. 执行更新
    return await self.project_repo.update(project_id, data.dict(exclude_unset=True))

删除操作 - 处理级联关系

async def delete_project(
    self,
    user_id: UUID,
    project_id: UUID,
    permanent: bool = False
) -> None:
    """删除项目 - 应用层级联处理"""
    
    # 1. 验证项目是否存在和权限
    project = await self.project_repo.get_by_id(project_id)
    if not project:
        raise NotFoundError("项目不存在")
    
    is_owner = await self.project_repo.check_user_permission(
        user_id, project_id, MemberRole.OWNER
    )
    if not is_owner:
        raise PermissionDeniedError("只有所有者可以删除项目")
    
    # 2. 使用事务处理级联删除
    async with self.db.begin():
        if permanent:
            # 永久删除：物理删除所有关联数据
            await self.project_member_repo.delete_by_project(project_id)
            await self.project_share_repo.delete_by_project(project_id)
            await self.project_snapshot_repo.delete_by_project(project_id)
            
            # 删除剧本及其关联数据
            screenplays = await self.screenplay_repo.get_by_project(project_id)
            for screenplay in screenplays:
                await self.delete_screenplay_cascade(screenplay.id)
            
            # 删除项目素材关联
            await self.project_resource_repo.delete_by_project(project_id)
            
            # 最后删除项目本身
            await self.project_repo.delete(project_id)
        else:
            # 软删除：标记删除时间
            await self.project_repo.soft_delete(project_id)
            await self.project_member_repo.soft_delete_by_project(project_id)
            await self.project_share_repo.soft_delete_by_project(project_id)

3. 后台任务 - 定期检查

使用 Celery 定期检查数据完整性：

# tasks/data_integrity.py
@shared_task
async def check_orphan_records():
    """检查孤儿记录并告警"""
    async with get_db() as db:
        # 检查项目表中的孤儿记录
        orphan_projects = await db.execute("""
            SELECT p.id, p.name, p.owner_id
            FROM projects p
            LEFT JOIN users u ON p.owner_id = u.user_id
            WHERE p.deleted_at IS NULL AND u.user_id IS NULL
        """)
        
        for project in orphan_projects:
            logger.error(f"发现孤儿项目: {project.id}, 所有者不存在: {project.owner_id}")
            # 发送告警通知
            await send_alert("data_integrity", f"项目 {project.name} 的所有者已被删除")

@shared_task
async def cleanup_soft_deleted_records():
    """清理超过 30 天的软删除记录"""
    async with get_db() as db:
        cutoff_date = datetime.utcnow() - timedelta(days=30)
        
        deleted_projects = await db.execute(
            select(Project).where(Project.deleted_at < cutoff_date)
        )
        
        for project in deleted_projects.scalars():
            await ProjectService(db).delete_project(
                user_id=project.owner_id,
                project_id=project.id,
                permanent=True
            )

常见关联关系处理模式

一对多关系（用户 → 项目）

# 删除父记录时的三种策略

# 策略 1: 禁止删除（如果有关联项目）
async def delete_user(self, user_id: UUID):
    project_count = await self.project_repo.count_by_owner(user_id)
    if project_count > 0:
        raise ValidationError("用户还有项目，无法删除")
    await self.user_repo.delete(user_id)

# 策略 2: 转移所有权
async def delete_user(self, user_id: UUID):
    await self.project_repo.transfer_ownership(
        from_user_id=user_id,
        to_user_id=admin_user_id
    )
    await self.user_repo.delete(user_id)

# 策略 3: 级联软删除
async def delete_user(self, user_id: UUID):
    await self.project_repo.soft_delete_by_owner(user_id)
    await self.user_repo.soft_delete(user_id)

多对多关系（项目 ↔ 成员）

# 添加关联时验证两端都存在
async def add_project_member(
    self,
    project_id: UUID,
    user_id: UUID,
    role: MemberRole
):
    if not await self.project_repo.exists(project_id):
        raise NotFoundError("项目不存在")
    
    if not await self.user_repo.exists(user_id):
        raise NotFoundError("用户不存在")
    
    # 检查是否已存在
    existing = await self.project_member_repo.get_member(project_id, user_id)
    if existing:
        raise ValidationError("用户已是项目成员")
    
    await self.project_member_repo.create({
        "project_id": project_id,
        "user_id": user_id,
        "role": role
    })

# 删除任一端时清理关联
async def delete_project(self, project_id: UUID):
    await self.project_member_repo.delete_by_project(project_id)
    await self.project_repo.delete(project_id)

自引用关系（文件夹树）

async def delete_folder(self, folder_id: UUID, user_id: UUID):
    """递归删除文件夹树"""
    async with self.db.begin():
        # 1. 获取所有子文件夹
        children = await self.folder_repo.get_children(folder_id)
        
        # 2. 递归删除子文件夹
        for child in children:
            await self.delete_folder(child.id, user_id)
        
        # 3. 处理文件夹内的项目（移到根目录）
        projects = await self.project_repo.get_by_folder(folder_id)
        for project in projects:
            await self.project_repo.update(project.id, {"folder_id": None})
        
        # 4. 删除文件夹本身
        await self.folder_repo.soft_delete(folder_id)

async def move_folder(self, folder_id: UUID, target_parent_id: Optional[UUID]):
    """移动文件夹 - 防止循环引用"""
    if target_parent_id:
        # 验证目标父文件夹存在
        if not await self.folder_repo.exists(target_parent_id):
            raise NotFoundError("目标文件夹不存在")
        
        # 防止循环引用：检查目标是否是当前文件夹的子孙
        is_descendant = await self.folder_repo.is_descendant(
            ancestor_id=folder_id,
            descendant_id=target_parent_id
        )
        if is_descendant:
            raise ValidationError("不能移动到自己的子文件夹")
    
    await self.folder_repo.update(folder_id, {"parent_folder_id": target_parent_id})

性能优化建议

1. 批量验证: 使用 IN 查询批量检查引用，避免 N+1 查询
2. 索引优化: 所有外键列必须创建索引（已在迁移脚本中处理）
3. 缓存策略: 对频繁访问的引用关系使用 Redis 缓存
4. 异步处理: 非关键路径的完整性检查放到后台任务
5. 事务控制: 使用数据库事务保证级联操作的原子性

# 批量验证示例
async def batch_move_projects(
    self,
    user_id: UUID,
    project_ids: list[UUID],
    target_folder_id: Optional[UUID]
):
    # 批量验证项目是否存在
    projects_exist = await self.project_repo.batch_exists(project_ids)
    missing_ids = [id for id, exists in projects_exist.items() if not exists]
    if missing_ids:
        raise NotFoundError(f"项目不存在: {missing_ids}")
    
    # 批量验证用户权限
    permissions = await self.project_repo.batch_check_user_permission(
        user_id, project_ids, MemberRole.EDITOR
    )
    unauthorized_ids = [id for id, has_perm in permissions.items() if not has_perm]
    if unauthorized_ids:
        raise PermissionDeniedError(f"无权限操作项目: {unauthorized_ids}")
    
    # 批量更新
    async with self.db.begin():
        await self.project_repo.batch_update(
            project_ids,
            {"folder_id": target_folder_id}
        )

实施要点

1. 所有 Service 方法都必须在操作前验证引用完整性
2. 删除操作必须明确处理级联关系（软删除或物理删除）
3. 批量操作使用批量验证方法优化性能
4. 事务边界要清晰，确保级联操作的原子性
5. 错误处理要明确，区分 NotFoundError 和 PermissionDeniedError
6. 后台任务定期检查数据完整性，及时发现和修复问题

完整的应用层引用完整性保证机制详见：

• 数据库设计规范 - 引用完整性保证

---

后续工作

文档维护建议

1. 新增服务文档时：

   • 参考 docs/requirements/backend/04-services/README.md 中的架构变更说明
   • 不要在表定义中添加物理外键约束
   • 不要在表定义中添加 UUID v7 数据库默认值

2. 更新现有文档时：

   • 检查是否包含过时的数据库定义
   • 及时清理不符合规范的内容

3. 代码实现时：

   • 以实际代码为准（server/app/models/）
   • 文档中的表定义仅供参考

---

相关文档

• UUID v7 生成迁移到应用层
• 移除数据库物理外键约束
• 服务文档 UUID 定义更新通知
• 数据库设计规范
• 服务文档总览 - 架构变更通知

---

总结

本次批量清理确保了所有服务设计文档与项目架构规范保持一致，移除了过时的数据库定义，为后续开发提供了准确的参考。所有引用完整性验证和 UUID 生成逻辑已统一在应用层实现。