多租户设计

// 错误示例：依赖应用层过滤
async function getSessions(tenantId: string) {
  return db.query('SELECT * FROM sessions WHERE tenant_id = $1', [tenantId]);
}
 
// 某天为了"快速排查"加了个 admin 接口
async function getAllSessions() {
  return db.query('SELECT * FROM sessions');  // 漏掉 tenant_id 过滤
}

-- 表级开启 RLS
-- ENABLE：让策略对普通用户生效（owner 默认豁免）
-- FORCE：让表的 owner 也受策略约束；如果应用用 owner 角色连接，少了 FORCE 等于没开
ALTER TABLE sessions ENABLE ROW LEVEL SECURITY;
ALTER TABLE sessions FORCE ROW LEVEL SECURITY;
 
-- 创建隔离策略
-- USING 作用于 SELECT/UPDATE/DELETE 的可见性
-- WITH CHECK 作用于 INSERT/UPDATE 的写入校验
-- 两者都设置，防止应用 bug 把行写到错误的 tenant_id
CREATE POLICY tenant_isolation ON sessions
  USING      (tenant_id = current_setting('app.current_tenant_id')::UUID)
  WITH CHECK (tenant_id = current_setting('app.current_tenant_id')::UUID);
 
-- 其他核心表同样处理
ALTER TABLE messages ENABLE ROW LEVEL SECURITY;
ALTER TABLE messages FORCE ROW LEVEL SECURITY;
CREATE POLICY tenant_isolation ON messages
  USING      (tenant_id = current_setting('app.current_tenant_id')::UUID)
  WITH CHECK (tenant_id = current_setting('app.current_tenant_id')::UUID);
 
ALTER TABLE knowledge_chunks ENABLE ROW LEVEL SECURITY;
ALTER TABLE knowledge_chunks FORCE ROW LEVEL SECURITY;
CREATE POLICY tenant_isolation ON knowledge_chunks
  USING      (tenant_id = current_setting('app.current_tenant_id')::UUID)
  WITH CHECK (tenant_id = current_setting('app.current_tenant_id')::UUID);

// 从连接池取出连接，设置租户上下文，然后执行业务查询
async function withTenantContext<T>(
  pool: Pool,
  tenantId: string,
  fn: (client: PoolClient) => Promise<T>
): Promise<T> {
  const client = await pool.connect();
  try {
    // SET LOCAL 的作用域是当前事务
    // 必须在事务内执行，确保连接归还连接池前变量被清除
    await client.query('BEGIN');
    await client.query(
      `SET LOCAL app.current_tenant_id = '${tenantId}'`
    );
    const result = await fn(client);
    await client.query('COMMIT');
    return result;
  } catch (err) {
    await client.query('ROLLBACK');
    throw err;
  } finally {
    client.release();
  }
}

CREATE TABLE tenant_configs (
  tenant_id UUID PRIMARY KEY REFERENCES tenants(id) ON DELETE CASCADE,
 
  -- Agent 行为配置
  system_prompt TEXT,                          -- 覆盖平台默认 system prompt
  allowed_skill_ids UUID[],                    -- Skill 白名单，NULL 表示允许全部
  knowledge_base_ids UUID[],                   -- 绑定的知识库 ID 列表
  max_context_turns INTEGER DEFAULT 20,        -- 最大对话轮数（影响 context window 用量）
 
  -- LLM 配置
  preferred_model VARCHAR(100),                -- 首选模型，NULL 使用平台默认
  fallback_model VARCHAR(100),                 -- 配额不足时的降级模型
  enable_builtin_knowledge BOOLEAN DEFAULT true,
 
  -- 合规与安全配置
  require_audit_log BOOLEAN DEFAULT false,     -- 是否记录完整审计日志（金融租户必开）
  pii_masking_enabled BOOLEAN DEFAULT false,   -- 敏感信息脱敏
  allowed_ip_ranges CIDR[],                    -- IP 白名单，NULL 不限制
 
  -- 元数据
  updated_at TIMESTAMPTZ DEFAULT NOW(),
  updated_by UUID                              -- 最后修改人
);

-- 检查某个 skill 是否在租户白名单内
SELECT allowed_skill_ids @> ARRAY[$1::UUID]
FROM tenant_configs
WHERE tenant_id = $2;

async getTenantConfig(tenantId: string): Promise<TenantConfig | null> {
  // L1
  const local = this.localCache.get(tenantId);
  if (local && local.expireAt > Date.now()) return local.config;
 
  // L2
  // Redis key 带花括号是 Redis Cluster hash tag 写法，
  // 同一租户的所有 key 落到同一个 slot，Lua 脚本和事务才能跨 key
  const cacheKey = `tenant:config:{${tenantId}}`;
  const cached = await this.redis.get(cacheKey);
  if (cached) {
    const config = this.deserializeConfig(cached);
    this.setLocalCache(tenantId, config);
    return config;
  }
 
  // L3：回源 Postgres，回填 L1 + L2
  const config = await this.loadFromDatabase(tenantId);
  if (!config) return null;
  await this.redis.setex(cacheKey, 300, this.serializeConfig(config));
  this.setLocalCache(tenantId, config);
  return config;
}
 
async updateTenantConfig(tenantId: string, patch: Partial<TenantConfig>) {
  // 1. 持久化写
  await this.writeToDatabase(tenantId, patch);
  // 2. 失效 Redis（其他进程 L2 miss 后会从 DB 拉到新值）
  await this.redis.del(`tenant:config:{${tenantId}}`);
  // 3. Pub/Sub 通知所有进程清 L1（生产多 worker 必需，单机空跑）
  await this.redis.publish('tenant:config:invalidate', tenantId);
  // 4. 当前进程的 L1 也立即清掉
  this.localCache.delete(tenantId);
}

this.redisSub.subscribe('tenant:config:invalidate');
this.redisSub.on('message', (channel, message) => {
  if (channel === 'tenant:config:invalidate') {
    this.localCache.delete(message);   // message 是 tenantId
  }
});

interface TenantQuota {
  tenantId: string;
  plan: 'starter' | 'pro' | 'enterprise';
 
  // LLM Token 配额
  dailyTokenLimit: number;      // 每日 Token 上限
  monthlyTokenLimit: number;    // 每月 Token 上限
 
  // 请求频率配额
  requestsPerMinute: number;    // 每分钟请求数
  requestsPerDay: number;       // 每日请求总数
 
  // 资源配额
  customSkillLimit: number;         // 自定义 Skill 数量上限
  knowledgeBaseSizeGb: number;      // 知识库总存储上限
  concurrentSessionLimit: number;   // 最大并发 session 数
}
 
// 不同套餐的配额预设
const PLAN_QUOTAS: Record<TenantQuota['plan'], Omit<TenantQuota, 'tenantId' | 'plan'>> = {
  starter: {
    dailyTokenLimit: 1_000_000,
    monthlyTokenLimit: 20_000_000,
    requestsPerMinute: 20,
    requestsPerDay: 5_000,
    customSkillLimit: 3,
    knowledgeBaseSizeGb: 1,
    concurrentSessionLimit: 50,
  },
  pro: {
    dailyTokenLimit: 10_000_000,
    monthlyTokenLimit: 200_000_000,
    requestsPerMinute: 100,
    requestsPerDay: 50_000,
    customSkillLimit: 20,
    knowledgeBaseSizeGb: 10,
    concurrentSessionLimit: 500,
  },
  enterprise: {
    dailyTokenLimit: 100_000_000,
    monthlyTokenLimit: 2_000_000_000,
    requestsPerMinute: 1_000,
    requestsPerDay: 500_000,
    customSkillLimit: 100,
    knowledgeBaseSizeGb: 100,
    concurrentSessionLimit: 5_000,
  },
};

// 检查今日 Token 配额是否充足
async function checkDailyTokenQuota(
  tenantId: string,
  estimatedTokens: number
): Promise<QuotaCheckResult> {
  const today = new Date().toISOString().slice(0, 10); // YYYY-MM-DD
  const usageKey = `quota:tokens:daily:{${tenantId}}:${today}`;
 
  const [used, limit] = await Promise.all([
    redis.get(usageKey).then(v => parseInt(v ?? '0', 10)),
    getQuotaLimit(tenantId, 'dailyTokenLimit'),
  ]);
 
  const remaining = limit - used;
 
  if (remaining < estimatedTokens) {
    // 计算配额重置时间（明天 00:00 UTC）
    const resetAt = getNextMidnightUTC();
    return {
      allowed: false,
      reason: 'daily_token_limit_exceeded',
      used,
      limit,
      resetAt,
    };
  }
 
  return { allowed: true, used, limit, remaining };
}
 
// 请求完成后扣减实际消耗的 Token
async function deductTokenUsage(
  tenantId: string,
  inputTokens: number,
  outputTokens: number,
  cacheHitTokens: number
): Promise<void> {
  const today = new Date().toISOString().slice(0, 10);
  const totalTokens = inputTokens + outputTokens;
 
  // 所有 key 共享 {tenantId} hash tag，在 Redis Cluster 下落到同一 slot，
  // pipeline 内的命令可以原子执行
  const dailyKey = `quota:tokens:daily:{${tenantId}}:${today}`;
  const monthKey = `quota:tokens:monthly:{${tenantId}}:${today.slice(0, 7)}`;
 
  // 用 pipeline 而不是 Promise.all：减少 RTT，并且让 INCRBY + EXPIRE
  // 在同一次往返中完成，避免极端情况下 INCRBY 后进程崩溃留下无过期时间的 key
  const pipe = redis.pipeline();
  pipe.incrby(dailyKey, totalTokens);
  pipe.expire(dailyKey, 25 * 3600);            // 25 小时，跨自然日有缓冲
  pipe.incrby(monthKey, totalTokens);
  pipe.expire(monthKey, 32 * 24 * 3600);       // 32 天，覆盖最长月份
  await pipe.exec();
 
  // 异步写入账单记录（非阻塞，写失败不影响主流程）
  recordBillingUsage(tenantId, inputTokens, outputTokens, cacheHitTokens).catch(
    err => logger.error({ err, tenantId }, 'billing record write failed')
  );
}

CREATE TABLE billing_records (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  tenant_id UUID NOT NULL REFERENCES tenants(id),
  period_start DATE NOT NULL,
  period_end DATE NOT NULL,
 
  -- Token 用量明细
  total_input_tokens BIGINT DEFAULT 0,
  total_output_tokens BIGINT DEFAULT 0,
  total_cache_hit_tokens BIGINT DEFAULT 0,   -- 命中缓存的 token（计费优惠）
  total_requests INTEGER DEFAULT 0,
 
  -- 费用明细（以分为单位）
  input_token_cost_cents INTEGER DEFAULT 0,
  output_token_cost_cents INTEGER DEFAULT 0,
  amount_cents INTEGER DEFAULT 0,            -- 总费用
  currency CHAR(3) DEFAULT 'USD',
 
  -- 账单状态
  -- draft：当月进行中，数据持续累积
  -- finalized：月末结算完毕，不再变更
  -- paid：已收款
  status VARCHAR(20) DEFAULT 'draft'
    CHECK (status IN ('draft', 'finalized', 'paid')),
 
  created_at TIMESTAMPTZ DEFAULT NOW(),
  updated_at TIMESTAMPTZ DEFAULT NOW()
);
 
-- 每个租户每个月只有一条 draft 记录
CREATE UNIQUE INDEX billing_records_tenant_period
  ON billing_records(tenant_id, period_start)
  WHERE status = 'draft';

local current_key = KEYS[1]
local previous_key = KEYS[2]
local limit = tonumber(ARGV[1])
local elapsed_ratio = tonumber(ARGV[2])   -- 当前分钟已过比例（0-1）
 
local current = tonumber(redis.call('GET', current_key) or '0')
local previous = tonumber(redis.call('GET', previous_key) or '0')
 
-- 滑动窗口估算：上一分钟剩余权重 + 当前分钟实际值
local estimated = math.floor(previous * (1 - elapsed_ratio)) + current
 
if estimated >= limit then
  return {0, estimated}
end
 
redis.call('INCR', current_key)
redis.call('EXPIRE', current_key, 120)
return {1, estimated + 1}

const currentMinute = Math.floor(Date.now() / 60_000);
const elapsedRatio = (Date.now() % 60_000) / 60_000;
 
// 两个 key 必须共享 {tenantId} hash tag，保证 Cluster 下落同 slot
const currentKey  = `ratelimit:{${tenantId}}:${currentMinute}`;
const previousKey = `ratelimit:{${tenantId}}:${currentMinute - 1}`;
 
const [allowed, estimated] = await redis.eval(
  SLIDING_WINDOW_SCRIPT,
  2, currentKey, previousKey,
  limitPerMinute.toString(), elapsedRatio.toString()
) as [number, number];

async function detectUsageAnomaly(
  tenantId: string,
  recentWindowTokens: number   // 最近 15 分钟的 Token 消耗
): Promise<void> {
  // 读取按 15 分钟窗口折算的 7 日基线
  const baselineKey = `usage:baseline15m:{${tenantId}}`;
  const baseline = await redis.get(baselineKey).then(v => parseFloat(v ?? '0'));
 
  if (baseline > 0 && recentWindowTokens > baseline * 3) {
    await alertingService.send({
      level: 'warning',
      tenantId,
      message: `异常用量检测：最近 15 分钟消耗 ${recentWindowTokens} Token，` +
               `7 日基线 ${baseline.toFixed(0)}，超出 3 倍阈值`,
    });
  }
}

// AgentFlow 请求处理入口，整合所有多租户组件
async function handleAgentRequest(
  request: FastifyRequest,
  reply: FastifyReply
): Promise<void> {
  const { tenantId } = request.tenant; // 由 JWT 中间件注入
  const { message, sessionId } = request.body as AgentRequestBody;
 
  // 1. 加载租户配置（Redis 缓存）
  const config = await configManager.get(tenantId);
 
  // 2. 检查请求速率
  const rateLimit = await quotaManager.checkRateLimit(
    tenantId,
    config.quota.requestsPerMinute
  );
  if (!rateLimit.allowed) {
    return reply.status(429).send({
      error: 'rate_limit_exceeded',
      retryAfter: rateLimit.retryAfter,
    });
  }
 
  // 3. 估算 Token 用量，检查日配额
  const estimatedTokens = estimateTokens(message, config.maxContextTurns);
  const quotaCheck = await quotaManager.checkDailyTokens(tenantId, estimatedTokens);
  if (!quotaCheck.allowed) {
    return reply.status(429).send({
      error: 'daily_quota_exceeded',
      resetAt: quotaCheck.resetAt,
    });
  }
 
  // 4. 在数据库事务中设置租户上下文，执行业务查询
  const result = await withTenantContext(dbPool, tenantId, async (client) => {
    // RLS 自动过滤，这里的查询只能看到当前租户的数据
    const session = await getOrCreateSession(client, sessionId, tenantId);
    const history = await getRecentMessages(client, sessionId, config.maxContextTurns);
    const knowledgeChunks = await searchKnowledge(
      client,
      message,
      config.knowledgeBaseIds
    );
    return { session, history, knowledgeChunks };
  });
 
  // 5. 调用 LLM
  const llmResponse = await llmGateway.complete({
    tenantId,
    model: config.preferredModel ?? DEFAULT_MODEL,
    systemPrompt: config.systemPrompt,
    messages: buildMessages(result.history, result.knowledgeChunks, message),
  });
 
  // 6. 扣减配额，写入账单记录
  await quotaManager.deductTokens(tenantId, llmResponse.usage);
 
  // 7. 写入会话记录（RLS 保证写入正确的 tenant）
  await withTenantContext(dbPool, tenantId, async (client) => {
    await saveMessage(client, sessionId, tenantId, message, llmResponse.content);
 
    // 金融租户需要完整审计日志
    if (config.requireAuditLog) {
      await writeAuditLog(client, tenantId, {
        requestId: request.id,
        input: message,
        output: llmResponse.content,
        model: llmResponse.model,
        usage: llmResponse.usage,
      });
    }
  });
 
  return reply.send({ content: llmResponse.content });
}