# 服务设计文档 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)`

**示例**：

```sql
-- ❌ 修改前
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()")`

**示例**：

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

-- ✅ 修改后
user_id UUID PRIMARY KEY,
```

```python
# ❌ 修改前
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. 其他包含过时定义的服务文档

---

## 验证结果

### 验证命令

```bash
# 验证不再包含物理外键约束
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 层 - 基础验证

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

```python
# 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 层 - 业务验证

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

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

```python
# 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
```

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

```python
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))
```

##### 删除操作 - 处理级联关系

```python
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 定期检查数据完整性：

```python
# 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
            )
```

### 常见关联关系处理模式

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

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

# 策略 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)
```

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

```python
# 添加关联时验证两端都存在
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)
```

#### 自引用关系（文件夹树）

```python
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. **事务控制**: 使用数据库事务保证级联操作的原子性

```python
# 批量验证示例
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. **后台任务**定期检查数据完整性，及时发现和修复问题

完整的应用层引用完整性保证机制详见：
- [数据库设计规范 - 引用完整性保证](./../../../.claude/skills/jointo-tech-stack/references/database.md#引用完整性保证)

---

## 后续工作

### 文档维护建议

1. **新增服务文档时**：
   - 参考 `docs/requirements/backend/04-services/README.md` 中的架构变更说明
   - 不要在表定义中添加物理外键约束
   - 不要在表定义中添加 UUID v7 数据库默认值

2. **更新现有文档时**：
   - 检查是否包含过时的数据库定义
   - 及时清理不符合规范的内容

3. **代码实现时**：
   - 以实际代码为准（`server/app/models/`）
   - 文档中的表定义仅供参考

---

## 相关文档

- [UUID v7 生成迁移到应用层](./2026-01-27-uuid-generation-app-layer.md)
- [移除数据库物理外键约束](./2026-01-27-remove-physical-foreign-keys.md)
- [服务文档 UUID 定义更新通知](./2026-01-27-service-docs-uuid-notice.md)
- [数据库设计规范](../../../.claude/skills/jointo-tech-stack/references/database.md)
- [服务文档总览 - 架构变更通知](../../../docs/requirements/backend/04-services/README.md#⚠️-重要架构变更通知)

---

## 总结

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