从零构建全栈技术知识库平台：Monorepo + Next.js 14 + FastAPI 实践

一套面向中小团队的全栈技术知识库：前端用 Turborepo 管理的 Next.js 14 多应用 Monorepo，后端用 FastAPI 提供文档/内容/鉴权 API，前后端通过 Admin 应用做 BFF 与身份打通。本文所有代码片段均取自真实仓库，可直接对照阅读。

项目背景与定位

做知识库类产品的痛点往往是：内容散落、前端多端重复、后端接口与鉴权各写一套。这个项目的定位是用一套 Monorepo 把"展示站 + 管理后台 + API 服务"收拢到同一个工程里，共享类型、组件与工具，降低协作成本。

项目定位

• 展示站（web）：面向访客的技术文档 / 面试题库浏览，支持国际化与 MDX / Markdown 渲染。
• 管理后台（admin）：内容管理与对外只读接口的承载方，同时作为 BFF 反向代理到 FastAPI。
• API 服务（fastapi-web）：商品/服务、个人名片、预约、文档埋点、受保护文件分发等接口，统一 JWT 鉴权。

技术选型与架构设计

为什么选择 Monorepo？

• 共享代码零成本：UI 组件、类型定义、工具函数、ESLint 配置都抽成 packages/*，应用层用 workspace:* 直接引用。
• 统一构建编排：Turborepo 负责任务依赖与缓存，一次 turbo run build 按 dependsOn 拓扑顺序构建所有应用与包。
• 依赖版本收敛：pnpm catalog: 把版本集中在 pnpm-workspace.yaml，避免"同一个库装了三个版本"。

核心技术栈

层	技术
构建编排	Turborepo 2.x + pnpm workspaces + pnpm catalog
前端	Next.js 14（App Router / RSC）、TypeScript、Tailwind CSS、styled-components
组件库	自研 packages/ui（React + CVA + clsx）
文档渲染	next-mdx-remote / react-markdown + remark-gfm + rehype-highlight
国际化	next-intl
后端	FastAPI + SQLAlchemy + MySQL + Redis
鉴权	JWT（jose）+ 密码哈希（passlib pbkdf2）+ NextAuth 桥接
部署	前端 Vercel，后端 Docker + host 网络模式

项目架构

┌─────────────┐      ┌──────────────┐      ┌──────────────────┐
│  web (访客)  │─────▶│ admin (BFF)  │─────▶│  fastapi-web     │
│ Next.js 14  │      │ Next.js 14   │◀────│  FastAPI         │
└─────────────┘      └──────┬───────┘      └──────────────────┘
                            │                     ▲
                   next-auth │                     │ JWT / httpOnly cookie
                            ▼                     │
                     MySQL / Redis ◀──────────────┘

前端两个应用 apps/web、apps/admin 共享 packages/ui、packages/types、packages/utils、packages/constants、packages/config、packages/api-client。

项目结构

interview-monorepo/
├── apps/
│   ├── web/          # 展示站（文档知识库）
│   └── admin/        # 管理后台 / BFF
├── packages/
│   ├── ui/           # React 组件库（Button / Card / Input）
│   ├── types/        # 跨应用类型
│   ├── utils/        # 工具函数
│   ├── constants/    # 常量
│   ├── config/       # 配置
│   └── api-client/   # 请求封装
├── turbo.json
└── pnpm-workspace.yaml

根 package.json 用 workspaces 声明应用与包，脚本统一委托给 turbo：

{
  "name": "interview-monorepo",
  "private": true,
  "packageManager": "pnpm@10.0.0",
  "workspaces": ["apps/*", "packages/*"],
  "scripts": {
    "build": "turbo run build",
    "dev": "turbo run dev",
    "lint": "turbo run lint",
    "type-check": "turbo run type-check"
  }
}

核心功能模块

1. Web 应用（前端知识库）

展示站的核心是"文档系统"：从本地 Markdown 或 Admin 只读接口拉取内容，在 RSC 中渲染。关键逻辑集中在 lib/docs.ts，采用三层数据源策略——构建/开发优先本地 docs/*.md，生产优先 Admin 的 /api/docs-public，任一层失败自动降级，保证构建不被外部服务阻塞。

// apps/web/src/lib/docs.ts（节选）
import fs from 'fs';
import path from 'path';

export interface Doc {
  slug: string;
  title: string;
  description?: string;
}

const DOCS_DIR = path.join(process.cwd(), '../../docs');

// Admin API 配置（公开接口，无需认证）
const ADMIN_API_URL = process.env.NEXT_PUBLIC_ADMIN_URL || 'http://localhost:3003';
const DOCS_API_ENDPOINT = `${ADMIN_API_URL}/api/docs-public`;

const isBuildTime = process.env.NEXT_PHASE === 'phase-production-build' ||
                    process.env.NEXT_PHASE === 'phase-development-build';
const isProduction = process.env.NODE_ENV === 'production' || process.env.VERCEL === '1';

// 构建时/开发优先本地文件；生产优先 Admin API，失败降级本地
export async function getAllDocs(): Promise<Doc[]> {
  if (isBuildTime || !isProduction) {
    const localDocs = getLocalDocs();
    if (localDocs.length > 0) return localDocs;
    const adminDocs = await fetchDocsFromAdmin();
    return adminDocs.length > 0 ? adminDocs : [];
  }
  const adminDocs = await fetchDocsFromAdmin();
  if (adminDocs.length > 0) return adminDocs;
  return getLocalDocs();
}

async function fetchDocsFromAdmin(): Promise<Doc[]> {
  try {
    const response = await fetch(DOCS_API_ENDPOINT, {
      cache: 'no-store',
      headers: { Referer: SITE_URL, Origin: SITE_URL },
    });
    if (!response.ok) return [];
    const data = await response.json();
    return data.success ? data.docs : [];
  } catch {
    return [];
  }
}

文档列表页是典型的 RSC 用法：async 组件直接 await getAllDocs()，force-dynamic 关闭静态化：

// apps/web/src/app/[locale]/docs/page.tsx（节选）
import Link from 'next/link';
import { getAllDocs, type Doc } from '@/lib/docs';

export const dynamic = 'force-dynamic';

export default async function DocsPage() {
  const docs = await getAllDocs();
  return (
    <div className="min-h-screen bg-gradient-to-br from-blue-50 to-indigo-100">
      <div className="space-y-3">
        {docs.map((doc) => (
          <Link
            key={doc.slug}
            href={`/docs/${doc.slug}`}
            className="block bg-white rounded-lg shadow-sm hover:shadow-md transition-shadow p-5"
          >
            <h3 className="text-lg font-semibold">{doc.title}</h3>
            {doc.description && <p className="text-gray-600 text-sm">{doc.description}</p>}
          </Link>
        ))}
      </div>
    </div>
  );
}

2. 共享组件库（packages/ui）

组件库用 class-variance-authority + clsx + tailwind-merge 做变体管理，main/types 直接指向 src/index.ts（源码直出，无需预编译产物），并用 peerDependencies 避免 React 多实例：

// packages/ui/src/components/Button.tsx
import React from "react";

export interface ButtonProps extends React.ButtonHTMLAttributes<HTMLButtonElement> {
  variant?: "default" | "destructive" | "outline" | "secondary" | "ghost" | "link";
  size?: "default" | "sm" | "lg" | "icon";
  children: React.ReactNode;
}

export const Button: React.FC<ButtonProps> = ({
  variant = "default", size = "default", className = "", children, ...props
}) => {
  const base = "inline-flex items-center justify-center rounded-md font-medium transition-colors focus-visible:outline-none disabled:opacity-50";
  const variants = {
    default: "bg-blue-600 text-white hover:bg-blue-700",
    outline: "border border-gray-300 bg-transparent hover:bg-gray-50",
    // ...
  } as const;
  const sizes = { default: "h-10 px-4 py-2", sm: "h-8 px-3 text-sm", lg: "h-12 px-6 text-lg" } as const;
  const classes = `${base} ${variants[variant]} ${sizes[size]} ${className}`;
  return <button className={classes} {...props}>{children}</button>;
};

3. Admin 应用（技术演示平台 / BFF）

Admin 同时承担三件事：NextAuth 认证、文档管理与对外只读接口（/api/admin/docs、/api/docs-public）、以及把 /api/fastapi/[...path] 反向代理到 FastAPI。这样前端只需和一个同源后端打交道，跨域与鉴权都在 admin 内收敛。

FastAPI 后端服务

后端用经典的"应用工厂 + 路由器 + 依赖注入"分层。create_app() 负责装配中间件、异常处理器、路由与生命周期，main.py 仅做入口：

# app/main.py
from .factory import create_app
from .config import settings
import uvicorn

app = create_app()

if __name__ == '__main__':
    uvicorn.run(
        "app.main:app",
        host=settings.host,
        port=settings.port,
        reload=settings.debug,
        log_level=settings.log_level,
    )

# app/factory.py（节选）
def create_app() -> FastAPI:
    """创建 FastAPI 应用实例"""
    models.Base.metadata.create_all(bind=engine)

    app = FastAPI(
        title=settings.app_name,
        description=settings.app_description,
        version=settings.app_version,
        docs_url=None,    # 禁用默认 docs，使用自定义路由
        redoc_url=None,
        openapi_url=settings.openapi_url,
        debug=settings.debug,
    )

    # 中间件顺序很重要：安全头最先，其次 IP 过滤，再路径保护，最后 CORS/限流
    setup_security_headers(app)
    setup_ip_filter(app)
    setup_path_protection(app)
    setup_middleware(app)
    setup_exception_handlers(app)

    @app.on_event("startup")
    async def startup_event():
        await redis_client.connect()

    @app.on_event("shutdown")
    async def shutdown_event():
        await redis_client.disconnect()

    app.include_router(system.router)
    app.include_router(auth.router)
    app.include_router(items.router)
    app.include_router(redis.router)
    app.include_router(doc_logs.router)
    app.include_router(profile.router)
    app.include_router(bookings.router)
    app.include_router(protected_files.router)
    return app

路由设计：读公开、写鉴权

以商品（服务）路由为例，读接口匿名可访问，写接口统一 Depends(get_admin_user)；分页参数用 Query(..., ge=, le=) 做边界约束：

# app/routers/items.py
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.orm import Session
from typing import List, Optional
from .. import crud, schemas
from ..database import get_db
from ..security import get_current_user, get_admin_user

router = APIRouter(prefix="/items", tags=["商品管理"],
                   responses={404: {"description": "商品未找到"}})

@router.get("/", response_model=List[schemas.Item])
def read_items(
    skip: int = Query(0, ge=0, description="跳过的记录数"),
    limit: int = Query(10, ge=1, le=100, description="返回的记录数"),
    db: Session = Depends(get_db),
    current_user: Optional[dict] = Depends(lambda: None),  # 公开访问
):
    """获取商品列表（公开访问）"""
    return crud.get_items(db, skip=skip, limit=limit)

@router.post("/", response_model=schemas.Item, status_code=201)
def create_item(
    item: schemas.ItemCreate,
    db: Session = Depends(get_db),
    admin_user: dict = Depends(get_admin_user),  # 需要管理员权限
):
    """创建新商品"""
    return crud.create_item(db=db, item=item)

@router.delete("/{item_id}")
def delete_item(
    item_id: int,
    db: Session = Depends(get_db),
    admin_user: dict = Depends(get_admin_user),
):
    """删除商品"""
    if not crud.delete_item(db, item_id=item_id):
        raise HTTPException(status_code=404, detail="商品未找到")
    return {"message": "商品删除成功"}

数据模型（Pydantic）

写接口用 ItemCreate/ItemUpdate，读接口用带 id 与时间的 Item，统一 from_attributes = True（Pydantic v2）支持 ORM 对象直接序列化：

# app/schemas.py（节选）
from pydantic import BaseModel, Field
from typing import Union, Optional
from datetime import datetime

class ItemBase(BaseModel):
    name: str = Field(..., min_length=1, max_length=100, description="商品名称")
    price: float = Field(..., gt=0, description="商品价格，必须大于0")
    is_offer: Union[bool, None] = Field(default=None, description="是否为特价商品")
    description: Optional[str] = Field(None, max_length=1000)

class ItemCreate(ItemBase):
    pass

class ItemUpdate(BaseModel):
    name: Optional[str] = Field(None, min_length=1, max_length=100)
    price: Optional[float] = Field(None, gt=0)

class Item(ItemBase):
    id: int
    created_at: Optional[datetime] = None
    updated_at: Optional[datetime] = None

    class Config:
        from_attributes = True  # Pydantic v2 语法

安全机制：多层级防护

鉴权用 jose 签发/校验 JWT，密码哈希用 passlib 的 pbkdf2_sha256（规避 bcrypt 的若干坑）。get_current_user 同时支持 Bearer Token 与 httpOnly Cookie 两种方式，get_admin_user 在前者基础上检查 role == "admin"：

# app/security.py（节选）
async def get_current_user(
    request: Request,
    credentials: Optional[HTTPAuthorizationCredentials] = Depends(HTTPBearer(auto_error=False)),
):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED, detail="无效的认证凭据",
        headers={"WWW-Authenticate": "Bearer"},
    )
    token = await get_token_from_request(request, credentials)
    try:
        payload = jwt.decode(token, settings.secret_key, algorithms=[settings.algorithm])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception
    user = fake_users_db.get(username)
    if user is None:
        raise credentials_exception
    return user

async def get_admin_user(current_user: dict = Depends(get_current_user)):
    if current_user["role"] != "admin":
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="权限不足")
    return current_user

为了打通前端 NextAuth 与后端 FastAPI 的身份体系，后端还实现了用 NextAuth 的 JWS 换取本服务 token 的桥接：手工拆三段 → base64url 补 padding → 用 NEXTAUTH_SECRET 做 HMAC-SHA256 → 恒定时间比较 hmac.compare_digest 校验签名与 exp → 比对 NEXTAUTH_ADMIN_EMAILS 白名单后签发本服务 access_token。

中间件：CORS + 速率限制

setup_middleware 先做 CORS，强制把自有域名追加进允许来源（防止 env 覆盖默认值导致跨域失败），再挂上自研限流中间件。限流按路径/方法分档（登录 / 严格 / 默认），Redis 故障时 fail-open 不阻断业务，并返回标准的 X-RateLimit-* 与 Retry-After 头：

# app/middleware.py（节选）
class RateLimitMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        path = request.url.path
        if path in ["/health", "/ping", "/api/docs", "/redoc"]:
            return await call_next(request)

        client_ip = (request.headers.get("x-forwarded-for", "").split(",")[0].strip()
                     or request.headers.get("x-real-ip")
                     or (request.client.host if request.client else "unknown"))

        rate_config = self._get_rate_limit_config(path, request.method)
        rate_key = f"ratelimit:{path}:{client_ip}"

        try:
            current = await redis_client.get(rate_key)
            current = int(current) if current else 0
            if current >= rate_config["requests"]:
                return JSONResponse(status_code=429, content={"error": True, "message": "请求过于频繁"})
            await redis_client.set(rate_key, str(current + 1), rate_config["window"])
            response = await call_next(request)
            response.headers["X-RateLimit-Limit"] = str(rate_config["requests"])
            return response
        except Exception:
            return await call_next(request)  # Redis 出错时不限制

Redis 客户端：全局单例 + 全方法降级

所有 Redis 操作都包了一层"无连接返回零值"的兜底——Redis 挂了业务不挂。并提供通用的 @cache_result 装饰器：

# app/redis_client.py（节选）
class RedisClient:
    def __init__(self):
        self.redis_client: Optional[redis.Redis] = None

    async def connect(self):
        try:
            self.redis_client = redis.from_url(settings.redis_url, decode_responses=True,
                                               socket_connect_timeout=5, socket_timeout=5)
            await self.redis_client.ping()
        except Exception as e:
            print(f"Redis 连接失败: {e}")
            self.redis_client = None  # 降级

    async def get(self, key: str):
        if not self.redis_client:
            return None
        try:
            value = await self.redis_client.get(key)
            if value:
                try:
                    return json.loads(value)
                except json.JSONDecodeError:
                    return value
            return None
        except Exception as e:
            print(f"Redis GET 错误: {e}")
            return None

def cache_result(key_prefix: str, expire: int = 3600):
    def decorator(func):
        async def wrapper(*args, **kwargs):
            cache_key = f"{key_prefix}:{hash(str(args) + str(kwargs))}"
            cached = await redis_client.get(cache_key)
            if cached is not None:
                return cached
            result = await func(*args, **kwargs)
            await redis_client.set(cache_key, result, expire)
            return result
        return wrapper
    return decorator

redis_client = RedisClient()

技术亮点与最佳实践

1. 前端：Turborepo 构建优化

turbo.json 用 dependsOn: ["^build"] 保证先构建依赖包，并精确声明 outputs 让缓存命中：

{
  "$schema": "https://turbo.build/schema.json",
  "tasks": {
    "build": {
      "dependsOn": ["^build"],
      "outputs": [".next/**", "!.next/cache/**", "dist/**"]
    },
    "dev": { "cache": false, "persistent": true },
    "lint": { "dependsOn": ["^lint"] },
    "type-check": { "dependsOn": ["^type-check"] }
  }
}

2. 后端：FastAPI 分层架构

• 路由层（routers）：只管协议与权限，不写业务逻辑。
• CRUD 层：crud.py 封装 SQLAlchemy 会话操作。
• Schema 层：Pydantic 做请求校验与响应序列化。
• 依赖注入：Depends(get_db)、Depends(get_admin_user) 贯穿全局，测试时易于替换。

3. 安全机制要点

• 密码只存哈希，且只接受哈希格式（拒绝明文）。
• 写接口统一管理员依赖，读接口按需匿名。
• NextAuth ↔ JWT 桥接用恒定时间比较，避免时序侧信道。
• 中间件顺序固定：安全头 → IP 过滤 → 路径保护 → CORS/限流。

部署与运维

前端：Vercel

apps/web 与 apps/admin 分别在 Vercel 配置独立的 project，构建命令走 turbo run build --filter=@interview/web。环境变量（含 NEXT_PUBLIC_*）在 Vercel 控制台配置。

后端：Docker + host 网络

生产用 docker-compose.prod.yml，network_mode: host 直接复用宿主机网络访问同机 MySQL / Redis，healthcheck 探活 /health：

# docker-compose.prod.yml（节选）
services:
  app:
    build:
      context: .
      dockerfile: Dockerfile
      args:
        - PORT=${PORT:-8086}
    container_name: fastapi-web-app
    network_mode: host
    environment:
      - APP_ENV=${APP_ENV}
      - DEBUG=${DEBUG}
      - DATABASE_URL=${DATABASE_URL}
      - REDIS_URL=${REDIS_URL}
      - SECRET_KEY=${SECRET_KEY}
      - NEXTAUTH_SECRET=${NEXTAUTH_SECRET}
      - NEXTAUTH_ADMIN_EMAILS=${NEXTAUTH_ADMIN_EMAILS}
      - ADMIN_PASSWORD_HASH=${ADMIN_PASSWORD_HASH}
      - RATE_LIMIT_REQUESTS=${RATE_LIMIT_REQUESTS}
      - RATE_LIMIT_WINDOW=${RATE_LIMIT_WINDOW}
    volumes:
      - ./app:/app/app
      - ./logs:/app/logs
    restart: always
    healthcheck:
      test: ["CMD", "sh", "-c", "curl -f http://0.0.0.0:${PORT:-8086}/health || exit 1"]
      interval: 30s
      timeout: 10s
      retries: 3

数据库与缓存

MySQL 通过 SQLAlchemy 连接，create_all 幂等建表并补列；Redis 作为限流计数与缓存层，且全程降级友好。

项目收益与反思

收益

• Monorepo 让组件/类型/工具复用率显著提升，多端一致性有保障。
• 后端"读公开、写鉴权"的依赖注入范式，使权限逻辑集中、易于审计。
• 三层数据源 + Redis 降级，让文档系统在外部依赖抖动时仍可用。

反思与改进

• fake_users_db 目前是内存字典，生产应替换为真实用户表（或复用 NextAuth 的用户源）。
• 限流键用 path + ip 较粗，后续可结合用户维度做更细粒度控制。
• 文档渲染链的 remark/rehype 插件较多，需要约束白名单防止 XSS。

未来规划

• 把身份认证统一到同一套用户服务，消除 admin / fastapi 两套账户。
• 引入端到端类型安全（OpenAPI → 前端 SDK 自动生成）。
• 文档编辑在 Admin 内闭环，支持版本与审核。

总结

这套架构的价值不在于"用了多少新技术"，而在于用 Monorepo 把前端多端与后端服务收拢到一套工程里，用清晰的依赖注入与降级策略保证可用性与安全性。代码层面，lib/docs.ts 的三层数据源、security.py 的 JWT/NextAuth 桥接、middleware.py 的分档限流，都是可直接复用的真实实现。

项目地址

• 前端 Monorepo：github.com/erishen/interview
• 后端 API：github.com/erishen/fastapi-web

在线体验

• 展示站：https://web.erishen.cn
• 管理后台：https://admin.erishen.cn

技术栈

Turborepo · pnpm · Next.js 14 · TypeScript · Tailwind CSS · FastAPI · SQLAlchemy · MySQL · Redis · JWT · NextAuth · Docker

相关阅读

• 《用 shadcn 风格组件库统一多端 UI》
• 《基于 RAG 的智能问答应用实战》

源码导航

完整工程分两个仓库：erishen/interview（Next.js monorepo）与 erishen/fastapi-web（FastAPI 后端）。

前端 monorepo（interview）

文件	作用
turbo.json	Turborepo 任务编排：build / dev / lint 流水线
package.json	根 workspace，聚合 apps/* 与 packages/*
apps/web/package.json	web 应用依赖与脚本
packages/ui/package.json	共享 UI 包配置
packages/ui/src/components/Button.tsx	共享 Button 组件
packages/ui/src/components/Card.tsx	共享 Card 组件
apps/web/src/lib/docs.ts	文档数据层（RSC 下读取知识库内容）

后端（fastapi-web）

文件	作用
app/factory.py	create_app() 应用工厂，装配路由/中间件/CORS
app/main.py	入口，启动 uvicorn
app/routers/items.py	items 路由（CRUD + 鉴权依赖）
app/schemas.py	Pydantic 请求/响应模型
app/security.py	鉴权与限流逻辑
app/middleware.py	自定义中间件（日志/耗时）
docker-compose.prod.yml	生产编排（web + db + redis）

常见问题