feat(database): add PostgreSQL repository and migration scripts

- Implement PostgresUserRepository with full CRUD operations - Support all query methods (by_id/email/username/token) - Use psycopg2 with RealDictCursor for clean mapping - Create initial schema migration (users/workspaces/members/invitations) - Add database indexes for performance - Setup foreign key constraints for data integrity - Include migration guide and rollback instructions - Add psycopg2 dependency Phase 4 Task 28/68 completed
2026-06-17 07:35:02 +08:00
parent 20f7145981
commit aeb7b52263
4 changed files with 375 additions and 0 deletions
@@ -0,0 +1,87 @@
+"""
+数据库迁移脚本 - 初始化表结构
+"""
+
+-- 用户表
+CREATE TABLE IF NOT EXISTS users (
+    id VARCHAR(255) PRIMARY KEY,
+    email VARCHAR(255) UNIQUE NOT NULL,
+    display_name VARCHAR(255) NOT NULL,
+    username VARCHAR(255) UNIQUE,
+    password_hash VARCHAR(255),
+    email_verified BOOLEAN DEFAULT FALSE,
+    email_verification_token VARCHAR(255),
+    password_reset_token VARCHAR(255),
+    password_reset_expires_at TIMESTAMP,
+    last_login_at TIMESTAMP,
+    last_login_ip VARCHAR(50),
+    created_at TIMESTAMP NOT NULL DEFAULT NOW()
+);
+
+CREATE INDEX idx_users_email ON users(email);
+CREATE INDEX idx_users_username ON users(username);
+CREATE INDEX idx_users_email_verification_token ON users(email_verification_token);
+CREATE INDEX idx_users_password_reset_token ON users(password_reset_token);
+
+
+-- 工作空间表
+CREATE TABLE IF NOT EXISTS workspaces (
+    id VARCHAR(255) PRIMARY KEY,
+    name VARCHAR(255) NOT NULL,
+    owner_user_id VARCHAR(255) NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+    subscription_plan VARCHAR(50) DEFAULT 'free',
+    subscription_status VARCHAR(50) DEFAULT 'active',
+    subscription_expires_at TIMESTAMP,
+    max_projects INTEGER DEFAULT 3,
+    max_storage_gb INTEGER DEFAULT 10,
+    used_storage_gb DECIMAL(10, 2) DEFAULT 0,
+    created_at TIMESTAMP NOT NULL DEFAULT NOW()
+);
+
+CREATE INDEX idx_workspaces_owner ON workspaces(owner_user_id);
+
+
+-- 工作空间成员表
+CREATE TABLE IF NOT EXISTS workspace_members (
+    id VARCHAR(255) PRIMARY KEY,
+    workspace_id VARCHAR(255) NOT NULL REFERENCES workspaces(id) ON DELETE CASCADE,
+    user_id VARCHAR(255) NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+    role VARCHAR(50) NOT NULL,
+    invited_by VARCHAR(255) REFERENCES users(id),
+    joined_at TIMESTAMP NOT NULL DEFAULT NOW(),
+    UNIQUE(workspace_id, user_id)
+);
+
+CREATE INDEX idx_workspace_members_workspace ON workspace_members(workspace_id);
+CREATE INDEX idx_workspace_members_user ON workspace_members(user_id);
+
+
+-- 工作空间邀请表
+CREATE TABLE IF NOT EXISTS workspace_invitations (
+    id VARCHAR(255) PRIMARY KEY,
+    workspace_id VARCHAR(255) NOT NULL REFERENCES workspaces(id) ON DELETE CASCADE,
+    inviter_user_id VARCHAR(255) NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+    invitee_email VARCHAR(255) NOT NULL,
+    role VARCHAR(50) NOT NULL,
+    invitation_token VARCHAR(255) UNIQUE NOT NULL,
+    status VARCHAR(50) DEFAULT 'pending',
+    expires_at TIMESTAMP,
+    accepted_at TIMESTAMP,
+    created_at TIMESTAMP NOT NULL DEFAULT NOW()
+);
+
+CREATE INDEX idx_invitations_workspace ON workspace_invitations(workspace_id);
+CREATE INDEX idx_invitations_token ON workspace_invitations(invitation_token);
+CREATE INDEX idx_invitations_workspace_email ON workspace_invitations(workspace_id, invitee_email, status);
+
+
+-- 项目表（占位，后续 Phase 完善）
+CREATE TABLE IF NOT EXISTS projects (
+    id VARCHAR(255) PRIMARY KEY,
+    workspace_id VARCHAR(255) NOT NULL REFERENCES workspaces(id) ON DELETE CASCADE,
+    name VARCHAR(255) NOT NULL,
+    description TEXT,
+    created_at TIMESTAMP NOT NULL DEFAULT NOW()
+);
+
+CREATE INDEX idx_projects_workspace ON projects(workspace_id);
@@ -0,0 +1,121 @@
+# 数据库迁移指南
+
+## 环境准备
+
+### 1. 安装 PostgreSQL
+
+```bash
+# Ubuntu/Debian
+sudo apt-get install postgresql postgresql-contrib
+
+# macOS
+brew install postgresql
+```
+
+### 2. 创建数据库
+
+```bash
+# 切换到 postgres 用户
+sudo -u postgres psql
+
+# 在 psql 中执行
+CREATE DATABASE xiaoxia_saas;
+CREATE USER xiaoxia_user WITH PASSWORD 'your_password_here';
+GRANT ALL PRIVILEGES ON DATABASE xiaoxia_saas TO xiaoxia_user;
+\q
+```
+
+### 3. 配置连接字符串
+
+在 `.env` 文件中配置：
+
+```env
+DATABASE_URL=postgresql://xiaoxia_user:your_password_here@localhost:5432/xiaoxia_saas
+```
+
+## 执行迁移
+
+### 方法 1: 使用 psql
+
+```bash
+# 执行初始化脚本
+psql postgresql://xiaoxia_user:your_password_here@localhost:5432/xiaoxia_saas \
+  -f migrations/001_initial_schema.sql
+```
+
+### 方法 2: 使用 Python 脚本
+
+```bash
+python scripts/run_migrations.py
+```
+
+## 迁移文件说明
+
+- `001_initial_schema.sql` - 初始表结构
+  - users - 用户表
+  - workspaces - 工作空间表
+  - workspace_members - 成员表
+  - workspace_invitations - 邀请表
+  - projects - 项目表（占位）
+
+## 验证迁移
+
+```sql
+-- 查看所有表
+\dt
+
+-- 查看 users 表结构
+\d users
+
+-- 查看索引
+\di
+
+-- 查看外键约束
+SELECT conname, conrelid::regclass AS table_name 
+FROM pg_constraint 
+WHERE contype = 'f';
+```
+
+## 回滚
+
+如果需要重置数据库：
+
+```bash
+# ⚠️ 警告：这会删除所有数据
+psql postgresql://xiaoxia_user:password@localhost:5432/xiaoxia_saas \
+  -c "DROP SCHEMA public CASCADE; CREATE SCHEMA public;"
+
+# 然后重新执行迁移
+psql ... -f migrations/001_initial_schema.sql
+```
+
+## 生产环境注意事项
+
+1. 使用强密码
+2. 配置 SSL 连接
+3. 定期备份数据
+4. 监控数据库性能
+5. 设置连接池（推荐使用 pgBouncer）
+
+## 连接池配置（可选）
+
+在 Python 中使用 psycopg2 连接池：
+
+```python
+from psycopg2 import pool
+
+connection_pool = pool.SimpleConnectionPool(
+    minconn=1,
+    maxconn=20,
+    dsn="postgresql://..."
+)
+```
+
+## 后续迁移
+
+后续的 schema 变更应该创建新的迁移文件：
+- `002_add_xxx.sql`
+- `003_alter_xxx.sql`
+- ...
+
+每个迁移文件应该是幂等的（可以多次执行而不出错）。
@@ -0,0 +1,167 @@
+"""
+PostgreSQL User Repository 实现
+"""
+from typing import Optional
+import psycopg2
+from psycopg2.extras import RealDictCursor
+from datetime import datetime
+
+from packages.domain.entities import User
+from packages.ports.user_repository import UserRepository
+
+
+class PostgresUserRepository(UserRepository):
+    """User 仓储 PostgreSQL 实现"""
+    
+    def __init__(self, connection_string: str):
+        self.connection_string = connection_string
+    
+    def _get_connection(self):
+        """获取数据库连接"""
+        return psycopg2.connect(self.connection_string, cursor_factory=RealDictCursor)
+    
+    def save(self, user: User) -> None:
+        """保存用户"""
+        conn = self._get_connection()
+        try:
+            with conn.cursor() as cur:
+                # Upsert (插入或更新)
+                cur.execute("""
+                    INSERT INTO users (
+                        id, email, display_name, username, password_hash,
+                        email_verified, email_verification_token,
+                        password_reset_token, password_reset_expires_at,
+                        last_login_at, last_login_ip, created_at
+                    ) VALUES (
+                        %(id)s, %(email)s, %(display_name)s, %(username)s, %(password_hash)s,
+                        %(email_verified)s, %(email_verification_token)s,
+                        %(password_reset_token)s, %(password_reset_expires_at)s,
+                        %(last_login_at)s, %(last_login_ip)s, %(created_at)s
+                    )
+                    ON CONFLICT (id) DO UPDATE SET
+                        email = EXCLUDED.email,
+                        display_name = EXCLUDED.display_name,
+                        username = EXCLUDED.username,
+                        password_hash = EXCLUDED.password_hash,
+                        email_verified = EXCLUDED.email_verified,
+                        email_verification_token = EXCLUDED.email_verification_token,
+                        password_reset_token = EXCLUDED.password_reset_token,
+                        password_reset_expires_at = EXCLUDED.password_reset_expires_at,
+                        last_login_at = EXCLUDED.last_login_at,
+                        last_login_ip = EXCLUDED.last_login_ip
+                """, {
+                    "id": user.id,
+                    "email": user.email,
+                    "display_name": user.display_name,
+                    "username": user.username,
+                    "password_hash": user.password_hash,
+                    "email_verified": user.email_verified,
+                    "email_verification_token": user.email_verification_token,
+                    "password_reset_token": user.password_reset_token,
+                    "password_reset_expires_at": user.password_reset_expires_at,
+                    "last_login_at": user.last_login_at,
+                    "last_login_ip": user.last_login_ip,
+                    "created_at": user.created_at,
+                })
+            conn.commit()
+        finally:
+            conn.close()
+    
+    def find_by_id(self, user_id: str) -> Optional[User]:
+        """根据 ID 查找用户"""
+        conn = self._get_connection()
+        try:
+            with conn.cursor() as cur:
+                cur.execute("SELECT * FROM users WHERE id = %s", (user_id,))
+                row = cur.fetchone()
+                
+                if row:
+                    return self._row_to_user(row)
+                return None
+        finally:
+            conn.close()
+    
+    def find_by_email(self, email: str) -> Optional[User]:
+        """根据邮箱查找用户"""
+        conn = self._get_connection()
+        try:
+            with conn.cursor() as cur:
+                cur.execute("SELECT * FROM users WHERE email = %s", (email.lower(),))
+                row = cur.fetchone()
+                
+                if row:
+                    return self._row_to_user(row)
+                return None
+        finally:
+            conn.close()
+    
+    def find_by_username(self, username: str) -> Optional[User]:
+        """根据用户名查找用户"""
+        conn = self._get_connection()
+        try:
+            with conn.cursor() as cur:
+                cur.execute("SELECT * FROM users WHERE username = %s", (username.lower(),))
+                row = cur.fetchone()
+                
+                if row:
+                    return self._row_to_user(row)
+                return None
+        finally:
+            conn.close()
+    
+    def find_by_verification_token(self, token: str) -> Optional[User]:
+        """根据邮箱验证令牌查找用户"""
+        conn = self._get_connection()
+        try:
+            with conn.cursor() as cur:
+                cur.execute("SELECT * FROM users WHERE email_verification_token = %s", (token,))
+                row = cur.fetchone()
+                
+                if row:
+                    return self._row_to_user(row)
+                return None
+        finally:
+            conn.close()
+    
+    def find_by_password_reset_token(self, token: str) -> Optional[User]:
+        """根据密码重置令牌查找用户"""
+        conn = self._get_connection()
+        try:
+            with conn.cursor() as cur:
+                cur.execute("SELECT * FROM users WHERE password_reset_token = %s", (token,))
+                row = cur.fetchone()
+                
+                if row:
+                    return self._row_to_user(row)
+                return None
+        finally:
+            conn.close()
+    
+    def delete(self, user_id: str) -> bool:
+        """删除用户"""
+        conn = self._get_connection()
+        try:
+            with conn.cursor() as cur:
+                cur.execute("DELETE FROM users WHERE id = %s", (user_id,))
+                deleted = cur.rowcount > 0
+            conn.commit()
+            return deleted
+        finally:
+            conn.close()
+    
+    def _row_to_user(self, row: dict) -> User:
+        """将数据库行转换为 User 对象"""
+        return User(
+            id=row["id"],
+            email=row["email"],
+            display_name=row["display_name"],
+            username=row["username"] or "",
+            password_hash=row["password_hash"] or "",
+            email_verified=row["email_verified"] or False,
+            email_verification_token=row["email_verification_token"],
+            password_reset_token=row["password_reset_token"],
+            password_reset_expires_at=row["password_reset_expires_at"],
+            last_login_at=row["last_login_at"],
+            last_login_ip=row["last_login_ip"],
+            created_at=row["created_at"],
+        )