🏗 ExitVideo-Bot · SaaS 多租户架构¶
版本：v1.0 · 作者：MiniMax · 2026-09-27
从"开源工具"转型"多租户 SaaS"的核心架构决策
目录¶
一、设计原则¶
商业 SaaS 的 7 条铁律¶
1. 多租户隔离是底线：租户数据绝不混
2. 计费透明可审计：每一笔消费都对得上
3. API 限流是基础：保护自己不被滥用
4. 异步任务可重入：任务失败必能重做
5. 数据保留有期限：不存 100 年
6. 权限细化到资源：RBAC 走到底
7. 监控覆盖每跳：N 个微服务 = N 个 trace
二、单租户 vs 多租户：决策¶
2.1 三种模式对比¶
| 模式 | 隔离强度 | 工程复杂度 | 适用规模 |
|---|---|---|---|
| 共享数据库 + tenant_id 列 | 🟡 软隔离 | 🟢 低 | 50-500 客户 |
| 共享数据库 + Row-Level Security | 🟢 强隔离 | 🟡 中 | 500+ |
| 独立数据库 per 客户 | 🟢🟢 极强隔离 | 🔴 高 | Enterprise |
2.2 推荐方案：RLS PostgreSQL¶
为什么不用"列加 tenant_id"？ - 开发阶段易漏写过滤条件 - 安全风险 = 数据库直连即泄露
为什么不用"独立数据库 per 客户"？ - 1 个客户 1 个数据库，运维成本爆炸（> 100 客户就崩溃） - 迁移、备份、连接池全是问题
推荐：PostgreSQL + Row-Level Security + 应用层 tenant context
-- 创建表时启用 RLS
CREATE TABLE accounts (
id BIGINT PRIMARY KEY,
tenant_id UUID NOT NULL REFERENCES tenants(id),
platform TEXT,
...
);
ALTER TABLE accounts ENABLE ROW LEVEL SECURITY;
CREATE POLICY tenant_isolation ON accounts
USING (tenant_id = current_setting('app.tenant_id')::UUID);
-- 每个请求开始时设置
SET LOCAL app.tenant_id = '<tenant_uuid>';
应用层：
# api/middleware/auth.py
async def set_tenant_context(db: AsyncSession, tenant_id: str):
await db.execute(f"SET LOCAL app.tenant_id = '{tenant_id}'")
yield # 请求周期内有效
# 自动结束，无需 tearDown
三、多租户数据隔离¶
3.1 关键实体 ER 图¶
┌──────────────┐ ┌──────────────┐ ┌──────────────┐
│ tenants │ 1─────* │ users │ 1─────* │ api_keys │
└──────┬───────┘ └──────┬───────┘ └──────────────┘
│ │
│ 1 │ 1
│ │
│ * │ *
┌──────▼───────┐ ┌──────▼───────┐
│ accounts │ 1─────* │ actions │
│ (视频账号) │ │ (审计日志) │
└──────┬───────┘ └──────────────┘
│ *
│
┌──────▼───────┐ ┌──────────────┐
│ devices │ │ invoices │
│ (注册的设备) │ │ (账单) │
└──────────────┘ └──────────────┘
3.2 表设计关键点¶
-- 租户（每个客户 1 个）
CREATE TABLE tenants (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
name TEXT NOT NULL,
plan TEXT NOT NULL DEFAULT 'starter', -- starter/growth/scale/enterprise
stripe_customer_id TEXT,
status TEXT NOT NULL DEFAULT 'active', -- active/trial/suspended/cancelled
trial_ends_at TIMESTAMPTZ,
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
-- 用户（每个租户内 N 个）
CREATE TABLE users (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
tenant_id UUID NOT NULL REFERENCES tenants(id),
email TEXT UNIQUE NOT NULL,
password_hash TEXT,
role TEXT NOT NULL DEFAULT 'member', -- owner/admin/member/viewer
mfa_enabled BOOLEAN DEFAULT false,
last_login_at TIMESTAMPTZ,
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
-- API Key（用于服务端调用）
CREATE TABLE api_keys (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
tenant_id UUID NOT NULL REFERENCES tenants(id),
name TEXT NOT NULL,
key_hash TEXT UNIQUE NOT NULL, -- bcrypt
scopes TEXT[], -- ['accounts:read', 'accounts:write', ...]
expires_at TIMESTAMPTZ,
last_used_at TIMESTAMPTZ,
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
-- 视频账号（核心资源）
CREATE TABLE accounts (
id BIGSERIAL PRIMARY KEY,
tenant_id UUID NOT NULL REFERENCES tenants(id),
platform TEXT NOT NULL,
country TEXT NOT NULL,
identifier TEXT NOT NULL, -- 用户名/Gmail
status TEXT NOT NULL DEFAULT 'active',
profile JSONB NOT NULL DEFAULT '{}', -- 设备指纹等敏感信息
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
ALTER TABLE accounts ENABLE ROW LEVEL SECURITY;
CREATE POLICY tenant_isolation ON accounts
USING (tenant_id = current_setting('app.tenant_id')::UUID);
-- 设备（每台真机）
CREATE TABLE devices (
id BIGSERIAL PRIMARY KEY,
tenant_id UUID NOT NULL REFERENCES tenants(id),
serial TEXT NOT NULL,
profile_index INT,
status TEXT NOT NULL, -- available/in_use/offline
last_heartbeat_at TIMESTAMPTZ,
UNIQUE(tenant_id, serial)
);
-- 审计日志（合规核心）
CREATE TABLE actions (
id BIGSERIAL PRIMARY KEY,
tenant_id UUID NOT NULL REFERENCES tenants(id),
user_id UUID,
api_key_id UUID,
resource_type TEXT NOT NULL, -- 'account' / 'device' / 'subscription'
resource_id BIGINT,
action TEXT NOT NULL, -- 'create'/'update'/'delete'/'login'
ip INET,
user_agent TEXT,
metadata JSONB DEFAULT '{}',
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE INDEX idx_actions_tenant_created ON actions(tenant_id, created_at DESC);
-- 账单（财务核心）
CREATE TABLE invoices (
id BIGSERIAL PRIMARY KEY,
tenant_id UUID NOT NULL REFERENCES tenants(id),
stripe_invoice_id TEXT UNIQUE,
amount_cents INT NOT NULL,
currency TEXT NOT NULL DEFAULT 'USD',
status TEXT NOT NULL, -- 'paid'/'open'/'void'
period_start TIMESTAMPTZ,
period_end TIMESTAMPTZ,
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
四、API Gateway 设计¶
4.1 端点设计¶
POST /v1/auth/register # 用户注册
POST /v1/auth/login # 登录，返回 JWT
POST /v1/auth/logout
POST /v1/auth/refresh # refresh token
GET /v1/me # 当前用户信息
PATCH /v1/me
POST /v1/me/mfa # 开启 2FA
# ===== 账号管理 =====
GET /v1/accounts # 列出账号
POST /v1/accounts # 创建账号（注册）
GET /v1/accounts/{id}
PATCH /v1/accounts/{id}
DELETE /v1/accounts/{id}
# ===== 设备管理 =====
GET /v1/devices
POST /v1/devices/register # 注册新设备
DELETE /v1/devices/{id}
# ===== 任务管理 =====
GET /v1/tasks # 任务队列状态
POST /v1/tasks/register # 注册任务
GET /v1/tasks/{id} # 任务详情 + 进度
POST /v1/tasks/{id}/cancel # 取消
# ===== 内容管理 =====
POST /v1/content/generate # AI 内容生成
POST /v1/content/publish # 发布任务
# ===== 计费 =====
GET /v1/billing/subscription
GET /v1/billing/usage
GET /v1/billing/invoices
# ===== 监控 =====
GET /v1/health/accounts # 账号健康
GET /v1/health/risk # 风控事件
# ===== Webhook =====
POST /v1/webhooks # 注册回调
GET /v1/webhooks
DELETE /v1/webhooks/{id}
4.2 中间件链¶
# api/middleware/chain.py
async def api_pipeline(request):
# 1. 签名验证（API Key 模式）
await verify_signature(request)
# 2. JWT 验证（用户模式）
user = await verify_jwt(request)
# 3. 租户上下文设置
tenant_id = user.tenant_id
async with db.session() as db:
await db.execute(f"SET LOCAL app.tenant_id = '{tenant_id}'")
# 设置后该连接的所有查询自动加 RLS 过滤
# 4. RBAC 权限检查
await check_scope(request, user, required_scopes)
# 5. 速率限制
await rate_limit(tenant_id, endpoint=request.url.path)
# 6. 审计日志
async with audit_action(request, user):
# 7. 业务路由
response = await route(request)
return response
4.3 API 鉴权三模式¶
| 模式 | 适用 | Header |
|---|---|---|
| JWT | 终端用户登录 | Authorization: Bearer <jwt> |
| API Key | 服务端集成 | X-ExitVideo-Key: ev_live_xxx |
| Webhook | 内部调用 | X-ExitVideo-Auth: <hmac_sha256> |
五、计费与配额¶
5.1 配额分类¶
| 维度 | Starter | Growth | Scale | Enterprise |
|---|---|---|---|---|
| 账号数 | 5 | 25 | 100 | 自定义 |
| 设备数 | 1 | 3 | 不限 | 不限 |
| API 调用/月 | 100 | 1,000 | 10,000 | 自定义 |
| 虚拟号采购/月 | 0（自带） | 5 | 30 | 自定义 |
| Webhook 事件/月 | 1,000 | 10,000 | 100,000 | 自定义 |
| 审计日志保留 | 7 天 | 30 天 | 365 天 | 永久 |
| 支持响应时间 | 24h | 4h | 1h | 15min |
| SLA | — | 99% | 99.5% | 99.9% |
| 价格/月 | $49 | $199 | $799 | 询价 |
5.2 超额计费¶
# api/billing/overage.py
OVERAGE_RATES = {
"extra_account": 2.50, # 每个超额的账号
"extra_api_call": 0.01, # 每 1000 次超额的 API 调用
"extra_virtual_number": 2.00, # 每个超额虚拟号
"extra_translation": 0.10, # 每篇翻译
}
async def check_and_charge_overage(tenant_id: str, action: str, count: int = 1):
"""操作前检查配额 + 操作后计算超额"""
usage = await get_current_usage(tenant_id)
limit = await get_plan_limits(tenant.tenant.plan)
if usage[action] + count > limit[action]:
# 超额，自动按 Stripe Add-on 计费
await stripe_invoice_item(
tenant.stripe_customer_id,
price=OVERAGE_RATES[action],
quantity=count,
description=f"{action} overage",
)
5.3 Stripe 集成¶
# api/billing/stripe.py
class StripeBilling:
def __init__(self, api_key: str):
self.stripe = stripe
self.stripe.api_key = api_key
async def create_subscription(self, tenant: Tenant, price_id: str):
return self.stripe.Subscription.create(
customer=tenant.stripe_customer_id,
items=[{"price": price_id}],
payment_behavior="default_incomplete",
expand=["latest_invoice.payment_intent"],
)
async def handle_webhook(self, payload: bytes, signature: str):
event = self.stripe.Webhook.construct_event(payload, signature, WEBHOOK_SECRET)
if event["type"] == "invoice.paid":
await mark_invoice_paid(event["data"]["object"])
elif event["type"] == "customer.subscription.deleted":
await suspend_tenant(event["data"]["object"]["customer"])
elif event["type"] == "customer.subscription.updated":
await update_plan(event["data"]["object"])
async def report_usage(self, subscription_id: str, action: str, qty: int):
"""上报用量给 Stripe 用于按需计费"""
self.stripe.SubscriptionItem.create_usage_record(
subscription_id,
quantity=qty,
timestamp=int(time.time()),
action="increment",
)
六、Worker 调度¶
6.1 任务分类¶
| 类型 | 例子 | SLA | 并发 |
|---|---|---|---|
| 实时任务 | 1 个账号注册 | 30s | 高 |
| 短批任务 | 设备健康巡检 | 60s | 中 |
| 长任务 | 内容生成 + 发布 | 5min | 低 |
| 周期任务 | 月度结算 | hours | 单跑 |
6.2 Celery vs RQ vs 自建¶
| Celery | RQ (Redis Queue) | Dramatiq | |
|---|---|---|---|
| 复杂度 | 🔴 高 | 🟢 低 | 🟡 中 |
| 多 Broker | ✅ | ❌ Redis only | ✅ |
| 监控 | 🟡 Flower 需部署 | 🟢 rq-dashboard | 🟡 自带 |
| 推荐度 | ⭐⭐⭐⭐⭐ | ⭐⭐⭐ | ⭐⭐⭐⭐ |
推荐：Celery (主) + RQ (轻量辅助)
6.3 任务定义¶
# api/tasks/register_account.py
@celery.task(bind=True, max_retries=3, default_retry_delay=60)
def register_account(self, tenant_id: str, account_id: int):
tenant = get_tenant(tenant_id)
device = get_available_device(tenant_id)
if not device:
raise self.retry(exc=Exception("No available device"))
try:
adb = ADBTool(device.serial)
agent = AutoGLMAgent()
sms = SMSProvider()
# 复用已有 client + audit
with audit_action("register", tenant_id, account_id):
registration_flow(adb, agent, sms, account_id)
update_account_status(account_id, "active")
notify_tenant(tenant_id, "register.success", account_id)
except Exception as exc:
update_account_status(account_id, "failed")
notify_tenant(tenant_id, "register.failed", account_id, str(exc))
raise self.retry(exc=exc)
七、关键技术决策¶
7.1 必须做的¶
| 决策 | 选项 A | 选项 B | 选择 |
|---|---|---|---|
| HTTP 框架 | FastAPI | Flask | FastAPI（自动 OpenAPI + 类型） |
| 数据库 | PostgreSQL | MongoDB | PostgreSQL（RLS 是杀手特性） |
| 任务队列 | Celery | RQ | Celery（更成熟） |
| 认证 | Auth0 | Supabase | 自建 + JWT（成本 + 控制） |
| 支付 | Stripe | Paddle | Stripe（API 设计更好） |
| 对象存储 | S3 | MinIO | S3（无所谓，自带 R2 更便宜） |
| 消息 | SMTP | Postmark | Postmark（送达率高） |
| 监控 | DataDog | Grafana Cloud | Grafana Cloud（便宜 + 自定义） |
| 日志 | ELK | Loki | Loki（免费 + 易上手） |
| 错误追踪 | Sentry | Rollbar | Sentry（Python SDK 优秀） |
7.2 永远不要做的¶
- ❌ 自己写 OAuth
- ❌ 自己实现 RBAC（用现成的 casbin / spicedb）
- ❌ 自己实现审计日志（用专用工具）
- ❌ 自己实现支付集成（用 Stripe / PayPal / Paddle）
- ❌ 自己写 CRM 系统（用现成 CRM + webhook）
- ❌ 自己实现 SSO（用 WorkOS / Clerk）
八、安全基线¶
8.1 网络层¶
[Cloudflare WAF] → [API Gateway]
- DDoS 防护
- 速率限制（per IP + per token）
- 国家/地区黑名单
- WAF 规则（SQL 注入 / XSS）
[API Gateway] → [Internal Service]
- JWT 校验
- RBAC 校验
- 审计日志
- 配额校验
8.2 数据层¶
| 项 | 方案 |
|---|---|
| 静态加密 | S3 KMS / PostgreSQL TDE |
| 传输加密 | TLS 1.3 only |
| 数据库凭证 | Hashicorp Vault |
| API Key 存储 | bcrypt（cost ≥ 12） |
| 密码存储 | Argon2id |
| 设备指纹存储 | AES-256-GCM（敏感信息） |
| 审计日志完整性 | 链式 hash（每行 hash 含上一行 hash） |
8.3 应用层¶
# api/security/checklist.py
SECURITY_REQUIREMENTS = [
"TLS-only", # 强制 HTTPS
"HSTS", # 1 年
"JWT-rotation", # JWT 24h 过期，Refresh 30 天
"CSRF-token", # 所有写操作需要
"Rate-limit-strict", # 100 req/min
"Audit-everything", # 全部写操作记录
"MFA-admin", # Admin 强制 2FA
"API-key-scope", # 每个 Key 限制 scope
"IP-allowlist", # Admin 限制 IP 段
"Secrets-vault", # 凭据走 Vault
"DB-RLS", # 数据库行级隔离
]
附录 A：MVP 启动清单¶
| 周 | 任务 | 工时 |
|---|---|---|
| 1 | FastAPI + JWT 认证 + 用户 CRUD | 40h |
| 2 | 租户 + RLS + 数据库 schema | 40h |
| 3 | accounts CRUD + 审计日志 | 40h |
| 4 | Stripe 订阅 + 配额计费 | 40h |
| 5 | Celery 任务队列集成 | 30h |
| 6 | 监控 + 日志 + 错误追踪 | 20h |
| 7-8 | Landing Page + 文档站 + Onboarding | 50h |
| 9-12 | 灰度发布 + 5 个种子客户 | 80h |
总计 12 周 / ~370 小时 = 一个全职 backend 工程师
附录 B：成本估算（Stage 2）¶
| 项 | 月成本 |
|---|---|
| Cloud 服务器 (2 workers + DB + Redis) | $200 |
| Sentry + Grafana + Loki | $50 |
| Stripe 抽成（假设 MRR $10K） | $300 |
| Cloudflare Pro | $20 |
| 域名 + 邮箱 + 杂项 | $30 |
| 合计 | ~$600 |
对应 MRR $10K，毛利率 94%。
维护者：MiniMax · 2026-09-27 · 配合
monetization/api/compliance/三大模块