diff --git a/.revision-progress b/.revision-progress
index 779bd9e..888f13c 100644
--- a/.revision-progress
+++ b/.revision-progress
@@ -9,3 +9,4 @@ phase4a done (caaec83)
 phase4 done (7a32d2c)
 phase5a done (frontend ts/vue/react/ssr)
 phase5b done (frontend vite/webpack/fe-testing-perf)
+phase5c done (backend mongodb/elasticsearch/data-pipeline/nodejs)
diff --git a/catalog.json b/catalog.json
index 6a2a9e1..6e4fb52 100644
--- a/catalog.json
+++ b/catalog.json
@@ -197,30 +197,34 @@
     {
       "stack": "data-pipeline",
       "domain": "backend",
-      "title": "数据管道与第三方接口对接",
+      "title": "数据管道基础",
       "index": "topics/backend/data-pipeline/index.md",
-      "lessons": 0,
-      "scenarios": 0,
+      "lessons": 1,
+      "scenarios": 1,
       "interview_questions": {
-        "基础": 0,
-        "进阶": 0,
-        "场景题": 0
+        "基础": 3,
+        "进阶": 3,
+        "场景题": 2
       },
-      "examples": []
+      "examples": [
+        "topics/backend/data-pipeline/examples/incremental-sync"
+      ]
     },
     {
       "stack": "elasticsearch",
       "domain": "backend",
       "title": "Elasticsearch",
       "index": "topics/backend/elasticsearch/index.md",
-      "lessons": 0,
-      "scenarios": 0,
+      "lessons": 1,
+      "scenarios": 1,
       "interview_questions": {
-        "基础": 0,
-        "进阶": 0,
-        "场景题": 0
+        "基础": 3,
+        "进阶": 3,
+        "场景题": 2
       },
-      "examples": []
+      "examples": [
+        "topics/backend/elasticsearch/examples/hybrid-search"
+      ]
     },
     {
       "stack": "gin",
@@ -290,14 +294,16 @@
       "domain": "backend",
       "title": "MongoDB",
       "index": "topics/backend/mongodb/index.md",
-      "lessons": 0,
-      "scenarios": 0,
+      "lessons": 1,
+      "scenarios": 1,
       "interview_questions": {
-        "基础": 0,
-        "进阶": 0,
-        "场景题": 0
+        "基础": 3,
+        "进阶": 3,
+        "场景题": 2
       },
-      "examples": []
+      "examples": [
+        "topics/backend/mongodb/examples/conversations"
+      ]
     },
     {
       "stack": "mq",
@@ -335,16 +341,18 @@
     {
       "stack": "nodejs",
       "domain": "backend",
-      "title": "Node.js 后端（Express / NestJS）",
+      "title": "Node.js（了解级）",
       "index": "topics/backend/nodejs/index.md",
-      "lessons": 0,
-      "scenarios": 0,
+      "lessons": 1,
+      "scenarios": 1,
       "interview_questions": {
-        "基础": 0,
-        "进阶": 0,
-        "场景题": 0
+        "基础": 3,
+        "进阶": 3,
+        "场景题": 2
       },
-      "examples": []
+      "examples": [
+        "topics/backend/nodejs/examples/express-bff"
+      ]
     },
     {
       "stack": "postgresql",
@@ -2872,13 +2880,18 @@
     },
     {
       "id": "backend.data-pipeline.index",
-      "title": "数据管道与第三方接口对接",
+      "title": "数据管道基础",
       "type": "index",
       "domain": "backend",
       "stack": [
         "data-pipeline"
       ],
-      "tags": [],
+      "tags": [
+        "ETL",
+        "增量同步",
+        "幂等",
+        "数据质量"
+      ],
       "level": null,
       "prerequisites": [],
       "related": [],
@@ -2888,7 +2901,139 @@
       ],
       "order": null,
       "path": "topics/backend/data-pipeline/index.md",
-      "sections": []
+      "sections": [
+        "学习路线",
+        "企业场景",
+        "面试题",
+        "示例代码",
+        "在 16 周计划中"
+      ]
+    },
+    {
+      "id": "backend.data-pipeline.interview",
+      "title": "数据管道面试题",
+      "type": "interview",
+      "domain": "backend",
+      "stack": [
+        "data-pipeline"
+      ],
+      "tags": [
+        "数据同步",
+        "幂等",
+        "重试",
+        "数据质量",
+        "面试"
+      ],
+      "level": "intermediate",
+      "prerequisites": [],
+      "related": [
+        "backend.data-pipeline.reliability"
+      ],
+      "est_hours": null,
+      "week": [
+        14,
+        16
+      ],
+      "order": null,
+      "path": "topics/backend/data-pipeline/interview.md",
+      "sections": [
+        "基础",
+        "进阶",
+        "场景题"
+      ],
+      "questions": {
+        "基础": 3,
+        "进阶": 3,
+        "场景题": 2
+      }
+    },
+    {
+      "id": "backend.data-pipeline.reliability",
+      "title": "数据管道基础：批处理与流处理、增量同步、限流重试、幂等、数据质量和告警",
+      "type": "lesson",
+      "domain": "backend",
+      "stack": [
+        "data-pipeline"
+      ],
+      "tags": [
+        "ETL",
+        "批处理",
+        "流处理",
+        "增量同步",
+        "水位线",
+        "限流",
+        "重试",
+        "幂等",
+        "死信",
+        "数据质量",
+        "告警"
+      ],
+      "level": "intermediate",
+      "prerequisites": [
+        "backend.postgresql.index-tx-mvcc",
+        "backend.mq.reliability"
+      ],
+      "related": [
+        "backend.data-pipeline.scenario.third-party-sync",
+        "backend.data-pipeline.interview",
+        "backend.elasticsearch.scenario.kb-search-sync"
+      ],
+      "est_hours": 3,
+      "week": [
+        14
+      ],
+      "order": 1,
+      "path": "topics/backend/data-pipeline/01-batch-stream-reliability.md",
+      "sections": [
+        "解释说明",
+        "案例",
+        "代码示例",
+        "易错点",
+        "面试常问"
+      ]
+    },
+    {
+      "id": "backend.data-pipeline.scenario.third-party-sync",
+      "title": "第三方平台订单同步：限流、断点续跑、迟到数据、对账和告警",
+      "type": "scenario",
+      "domain": "backend",
+      "stack": [
+        "data-pipeline",
+        "postgresql",
+        "redis",
+        "mq"
+      ],
+      "tags": [
+        "第三方对接",
+        "增量同步",
+        "限流",
+        "对账",
+        "迟到数据",
+        "告警"
+      ],
+      "level": "intermediate",
+      "prerequisites": [
+        "backend.data-pipeline.reliability"
+      ],
+      "related": [
+        "backend.data-pipeline.interview",
+        "backend.redis.lock-ratelimit"
+      ],
+      "est_hours": 2.5,
+      "week": [
+        14
+      ],
+      "order": null,
+      "path": "topics/backend/data-pipeline/scenarios/third-party-sync.md",
+      "sections": [
+        "业务背景与问题",
+        "约束",
+        "方案设计",
+        "关键代码",
+        "取舍",
+        "易错点",
+        "面试怎么讲"
+      ]
     },
     {
       "id": "backend.elasticsearch.index",
@@ -2898,7 +3043,12 @@
       "stack": [
         "elasticsearch"
       ],
-      "tags": [],
+      "tags": [
+        "Elasticsearch",
+        "全文检索",
+        "混合检索",
+        "向量检索"
+      ],
       "level": null,
       "prerequisites": [],
       "related": [],
@@ -2909,7 +3059,143 @@
       ],
       "order": null,
       "path": "topics/backend/elasticsearch/index.md",
-      "sections": []
+      "sections": [
+        "学习路线",
+        "企业场景",
+        "面试题",
+        "示例代码",
+        "在 16 周计划中",
+        "官方资料"
+      ]
+    },
+    {
+      "id": "backend.elasticsearch.interview",
+      "title": "Elasticsearch 面试题",
+      "type": "interview",
+      "domain": "backend",
+      "stack": [
+        "elasticsearch"
+      ],
+      "tags": [
+        "Elasticsearch",
+        "倒排索引",
+        "混合检索",
+        "面试"
+      ],
+      "level": "intermediate",
+      "prerequisites": [],
+      "related": [
+        "backend.elasticsearch.search-hybrid"
+      ],
+      "est_hours": null,
+      "week": [
+        14,
+        16
+      ],
+      "order": null,
+      "path": "topics/backend/elasticsearch/interview.md",
+      "sections": [
+        "基础",
+        "进阶",
+        "场景题"
+      ],
+      "questions": {
+        "基础": 3,
+        "进阶": 3,
+        "场景题": 2
+      }
+    },
+    {
+      "id": "backend.elasticsearch.scenario.kb-search-sync",
+      "title": "知识库混合检索：数据库到 ES 的同步、权限过滤、分词词典和检索效果评估",
+      "type": "scenario",
+      "domain": "backend",
+      "stack": [
+        "elasticsearch",
+        "postgresql",
+        "mq",
+        "rag"
+      ],
+      "tags": [
+        "混合检索",
+        "数据同步",
+        "CDC",
+        "outbox",
+        "权限过滤",
+        "IK 词典",
+        "召回率"
+      ],
+      "level": "intermediate",
+      "prerequisites": [
+        "backend.elasticsearch.search-hybrid",
+        "ai.rag.hybrid-rerank"
+      ],
+      "related": [
+        "backend.elasticsearch.interview",
+        "ai.rag.scenario.permission-aware-kb",
+        "backend.mq.reliability"
+      ],
+      "est_hours": 2.5,
+      "week": [
+        6,
+        14
+      ],
+      "order": null,
+      "path": "topics/backend/elasticsearch/scenarios/kb-search-sync.md",
+      "sections": [
+        "业务背景与问题",
+        "约束",
+        "方案设计",
+        "关键代码",
+        "取舍",
+        "易错点",
+        "面试怎么讲"
+      ]
+    },
+    {
+      "id": "backend.elasticsearch.search-hybrid",
+      "title": "Elasticsearch：倒排索引、映射与中文分词、BM25、kNN 向量检索、混合检索和别名重建",
+      "type": "lesson",
+      "domain": "backend",
+      "stack": [
+        "elasticsearch"
+      ],
+      "tags": [
+        "Elasticsearch",
+        "倒排索引",
+        "BM25",
+        "IK 分词",
+        "cjk",
+        "dense_vector",
+        "kNN",
+        "RRF",
+        "混合检索",
+        "别名",
+        "reindex"
+      ],
+      "level": "intermediate",
+      "prerequisites": [
+        "ai.rag.hybrid-rerank"
+      ],
+      "related": [
+        "backend.elasticsearch.scenario.kb-search-sync",
+        "ai.rag.index",
+        "backend.elasticsearch.interview"
+      ],
+      "est_hours": 3,
+      "week": [
+        6,
+        14
+      ],
+      "order": 1,
+      "path": "topics/backend/elasticsearch/01-search-basics-hybrid.md",
+      "sections": [
+        "解释说明",
+        "案例",
+        "代码示例",
+        "易错点",
+        "面试常问"
+      ]
     },
     {
       "id": "backend.gin.index",
@@ -3534,6 +3820,49 @@
       "path": "topics/backend/java-spring/index.md",
       "sections": []
     },
+    {
+      "id": "backend.mongodb.document-model",
+      "title": "MongoDB 基础：文档建模（嵌入还是引用）、索引与 ESR、聚合、事务和变更流",
+      "type": "lesson",
+      "domain": "backend",
+      "stack": [
+        "mongodb"
+      ],
+      "tags": [
+        "MongoDB",
+        "文档模型",
+        "嵌入",
+        "引用",
+        "复合索引",
+        "ESR",
+        "聚合管道",
+        "事务",
+        "变更流",
+        "副本集"
+      ],
+      "level": "basic",
+      "prerequisites": [
+        "backend.postgresql.index"
+      ],
+      "related": [
+        "backend.mongodb.scenario.conversation-store",
+        "backend.elasticsearch.index",
+        "backend.mongodb.interview"
+      ],
+      "est_hours": 3,
+      "week": [
+        14
+      ],
+      "order": 1,
+      "path": "topics/backend/mongodb/01-document-model.md",
+      "sections": [
+        "解释说明",
+        "案例",
+        "代码示例",
+        "易错点",
+        "面试常问"
+      ]
+    },
     {
       "id": "backend.mongodb.index",
       "title": "MongoDB",
@@ -3542,7 +3871,11 @@
       "stack": [
         "mongodb"
       ],
-      "tags": [],
+      "tags": [
+        "MongoDB",
+        "文档数据库",
+        "NoSQL"
+      ],
       "level": null,
       "prerequisites": [],
       "related": [],
@@ -3552,7 +3885,93 @@
       ],
       "order": null,
       "path": "topics/backend/mongodb/index.md",
-      "sections": []
+      "sections": [
+        "学习路线",
+        "企业场景",
+        "面试题",
+        "示例代码",
+        "在 16 周计划中",
+        "官方资料"
+      ]
+    },
+    {
+      "id": "backend.mongodb.interview",
+      "title": "MongoDB 面试题",
+      "type": "interview",
+      "domain": "backend",
+      "stack": [
+        "mongodb"
+      ],
+      "tags": [
+        "MongoDB",
+        "文档模型",
+        "索引",
+        "事务",
+        "面试"
+      ],
+      "level": "basic",
+      "prerequisites": [],
+      "related": [
+        "backend.mongodb.document-model"
+      ],
+      "est_hours": null,
+      "week": [
+        14,
+        16
+      ],
+      "order": null,
+      "path": "topics/backend/mongodb/interview.md",
+      "sections": [
+        "基础",
+        "进阶",
+        "场景题"
+      ],
+      "questions": {
+        "基础": 3,
+        "进阶": 3,
+        "场景题": 2
+      }
+    },
+    {
+      "id": "backend.mongodb.scenario.conversation-store",
+      "title": "AI 客服会话存储：海量对话记录的建模、查询、归档和同步",
+      "type": "scenario",
+      "domain": "backend",
+      "stack": [
+        "mongodb",
+        "postgresql",
+        "elasticsearch"
+      ],
+      "tags": [
+        "对话记录",
+        "文档建模",
+        "TTL",
+        "归档",
+        "变更流",
+        "多租户"
+      ],
+      "level": "intermediate",
+      "prerequisites": [
+        "backend.mongodb.document-model"
+      ],
+      "related": [
+        "backend.mongodb.interview"
+      ],
+      "est_hours": 2,
+      "week": [
+        14
+      ],
+      "order": null,
+      "path": "topics/backend/mongodb/scenarios/conversation-store.md",
+      "sections": [
+        "业务背景与问题",
+        "约束",
+        "方案设计",
+        "关键代码",
+        "取舍",
+        "易错点",
+        "面试怎么讲"
+      ]
     },
     {
       "id": "backend.mq.index",
@@ -4017,15 +4436,63 @@
         "面试常问"
       ]
     },
+    {
+      "id": "backend.nodejs.express-nest",
+      "title": "Node.js 后端（了解级）：事件循环、Express 5 中间件与错误处理、NestJS 的模块化思路、做 BFF",
+      "type": "lesson",
+      "domain": "backend",
+      "stack": [
+        "nodejs"
+      ],
+      "tags": [
+        "Node.js",
+        "事件循环",
+        "Express 5",
+        "中间件",
+        "错误处理",
+        "NestJS",
+        "依赖注入",
+        "BFF",
+        "SSE"
+      ],
+      "level": "basic",
+      "prerequisites": [
+        "frontend.typescript.type-system"
+      ],
+      "related": [
+        "frontend.ssr.nextjs-app-router",
+        "backend.nodejs.scenario.ai-bff",
+        "backend.nodejs.interview"
+      ],
+      "est_hours": 2.5,
+      "week": [
+        7,
+        14
+      ],
+      "order": 1,
+      "path": "topics/backend/nodejs/01-express-nest-basics.md",
+      "sections": [
+        "解释说明",
+        "案例",
+        "代码示例",
+        "易错点",
+        "面试常问"
+      ]
+    },
     {
       "id": "backend.nodejs.index",
-      "title": "Node.js 后端（Express / NestJS）",
+      "title": "Node.js（了解级）",
       "type": "index",
       "domain": "backend",
       "stack": [
         "nodejs"
       ],
-      "tags": [],
+      "tags": [
+        "Node.js",
+        "Express",
+        "NestJS",
+        "BFF"
+      ],
       "level": null,
       "prerequisites": [],
       "related": [],
@@ -4036,7 +4503,94 @@
       ],
       "order": null,
       "path": "topics/backend/nodejs/index.md",
-      "sections": []
+      "sections": [
+        "学习路线",
+        "企业场景",
+        "面试题",
+        "示例代码",
+        "在 16 周计划中"
+      ]
+    },
+    {
+      "id": "backend.nodejs.interview",
+      "title": "Node.js 面试题",
+      "type": "interview",
+      "domain": "backend",
+      "stack": [
+        "nodejs"
+      ],
+      "tags": [
+        "Node.js",
+        "事件循环",
+        "Express",
+        "NestJS",
+        "面试"
+      ],
+      "level": "basic",
+      "prerequisites": [],
+      "related": [
+        "backend.nodejs.express-nest"
+      ],
+      "est_hours": null,
+      "week": [
+        14,
+        16
+      ],
+      "order": null,
+      "path": "topics/backend/nodejs/interview.md",
+      "sections": [
+        "基础",
+        "进阶",
+        "场景题"
+      ],
+      "questions": {
+        "基础": 3,
+        "进阶": 3,
+        "场景题": 2
+      }
+    },
+    {
+      "id": "backend.nodejs.scenario.ai-bff",
+      "title": "用 Node.js 给 AI 应用做 BFF：鉴权、限流、流式转发、超时和断开处理",
+      "type": "scenario",
+      "domain": "backend",
+      "stack": [
+        "nodejs",
+        "redis",
+        "nginx",
+        "fastapi"
+      ],
+      "tags": [
+        "BFF",
+        "SSE",
+        "流式转发",
+        "限流",
+        "超时",
+        "request id"
+      ],
+      "level": "intermediate",
+      "prerequisites": [
+        "backend.nodejs.express-nest"
+      ],
+      "related": [
+        "backend.nodejs.interview",
+        "frontend.ssr.scenario.help-center"
+      ],
+      "est_hours": 2,
+      "week": [
+        14
+      ],
+      "order": null,
+      "path": "topics/backend/nodejs/scenarios/ai-bff.md",
+      "sections": [
+        "业务背景与问题",
+        "约束",
+        "方案设计",
+        "关键代码",
+        "取舍",
+        "易错点",
+        "面试怎么讲"
+      ]
     },
     {
       "id": "backend.postgresql.index",
diff --git a/topics/backend/data-pipeline/01-batch-stream-reliability.md b/topics/backend/data-pipeline/01-batch-stream-reliability.md
new file mode 100644
index 0000000..e2db65f
--- /dev/null
+++ b/topics/backend/data-pipeline/01-batch-stream-reliability.md
@@ -0,0 +1,100 @@
+---
+id: backend.data-pipeline.reliability
+title: 数据管道基础：批处理与流处理、增量同步、限流重试、幂等、数据质量和告警
+type: lesson
+domain: backend
+stack: [data-pipeline]
+tags: [ETL, 批处理, 流处理, 增量同步, 水位线, 限流, 重试, 幂等, 死信, 数据质量, 告警]
+level: intermediate
+prerequisites: [backend.postgresql.index-tx-mvcc, backend.mq.reliability]
+est_hours: 3
+related: [backend.data-pipeline.scenario.third-party-sync, backend.data-pipeline.interview, backend.elasticsearch.scenario.kb-search-sync]
+week: [14]
+order: 1
+---
+
+# 数据管道基础：批处理与流处理、增量同步、限流重试、幂等、数据质量和告警
+
+> BOSS直聘 2026-10-23 的调研中，部分 AI 应用岗位要求“数据处理 / 数据管道”能力：知识库文档的采集清洗、第三方平台数据对接、报表数据准备。这里讲后端工程师需要掌握的基础，不涉及大数据平台。
+
+## 解释说明
+
+### 批处理和流处理
+
+| | 批处理 | 流处理 |
+| --- | --- | --- |
+| 触发 | 定时（每小时 / 每天） | 事件到达即处理 |
+| 延迟 | 分钟到小时 | 秒级 |
+| 实现 | 定时任务 + SQL / Python | 消息队列（Kafka / RabbitMQ）+ 消费者，或 CDC |
+| 适合 | 报表、对账、全量重建 | 搜索同步、实时告警、实时推荐 |
+
+很多系统两者结合：流处理保证实时，批处理定期对账兜底。
+
+### 增量同步的关键
+
+1. **水位线**：记录上次同步到哪里（时间戳 + 主键打破平局），失败后从水位线继续。**水位线只在本批数据成功落库后推进**。
+2. **限流**：第三方接口有调用频率限制，用令牌桶控制；收到 429 时遵守 `Retry-After`。
+3. **重试**：区分可重试错误（超时、5xx、429）和不可重试错误（401、参数错误）；指数退避 + 抖动（避免大量客户端同时重试），有上限。
+4. **幂等**：重复运行结果一致——按主键 upsert，并且比较版本号 / 更新时间，旧数据不能覆盖新数据。
+5. **坏数据**：单条坏数据进入死信表，不阻塞整批；坏数据比例超过阈值说明上游出问题了，应停止并告警。
+6. **数据质量检查**：行数波动、空值率、取值范围、唯一性、与源头的对账。
+7. **可观测**：每次运行记录拉取数、写入数、死信数、耗时；失败和质量异常告警。
+
+### 常用工具（了解）
+
+定时调度：cron、Celery Beat、APScheduler；编排：Airflow、Dagster、Prefect；CDC：Debezium；数据质量：Great Expectations 一类工具。小团队先用“定时任务 + 本课的这些原则”就够了。
+
+## 案例
+
+每 10 分钟从电商平台开放接口同步订单到自己的数据库，供 AI 报表 Agent 查询：接口每秒限 50 次、偶尔超时；有时返回状态值异常的数据；同步中断后要能续跑；不能产生重复订单。
+
+## 代码示例
+
+[examples/incremental-sync](examples/incremental-sync/)（只用标准库，`python pipeline.py` 运行，`test_pipeline.py` 6 个测试通过）：
+
+```text
+第一次：拉取 251，写入 250，死信 1，接口调用 5 次（含 2 次失败重试）
+第二次（没有新数据）：拉取 0
+orders 表 250 行，水位线 ('2026-10-23T11:00:00', 'bad1')
+```
+
+幂等且防止旧版本覆盖：
+
+```sql
+INSERT INTO orders (id, amount_cents, status, updated_at) VALUES (:id, :amount_cents, :status, :updated_at)
+ON CONFLICT(id) DO UPDATE SET amount_cents = excluded.amount_cents, status = excluded.status, updated_at = excluded.updated_at
+WHERE excluded.updated_at > orders.updated_at
+```
+
+重试（指数退避 + 全抖动，遵守 Retry-After）：
+
+```python
+def with_retry(fn, *, attempts=5, base=0.2, cap=5.0, sleep=time.sleep, rng=None):
+    rng = rng or random.Random()
+    for i in range(attempts):
+        try:
+            return fn()
+        except RetryableError as e:
+            if i == attempts - 1:
+                raise
+            delay = e.retry_after if e.retry_after is not None else rng.uniform(0, min(cap, base * 2 ** i))
+            sleep(delay)
+```
+
+质量检查不通过时整批回滚、水位线不前进并告警（见 `test_quality_gate_rolls_back_and_alerts`）。
+
+## 易错点
+
+- 先推进水位线再写数据：中途失败就丢数据。
+- 只用时间戳做水位线：同一秒内有多条数据时会漏或重复；要加主键打破平局。
+- **迟到数据**：来源系统的事务晚提交，`updated_at` 比水位线还早，被永远跳过。办法：每次往回多拉一个重叠窗口（例如 5 分钟），依赖幂等写入去重；再加定期对账。
+- 对所有错误都重试，包括 401，结果被对方封禁。
+- 没有抖动的重试，故障恢复瞬间所有任务一起打过去。
+- 坏数据直接让整个任务失败，或者相反——静默丢弃没有任何记录。
+
+## 面试常问
+
+- 怎么设计一个可以断点续传的增量同步任务？
+- 怎么保证同步任务幂等？
+- 第三方接口限流和不稳定怎么处理？
+- 怎么发现数据质量问题？
diff --git a/topics/backend/data-pipeline/examples/incremental-sync/pipeline.py b/topics/backend/data-pipeline/examples/incremental-sync/pipeline.py
new file mode 100644
index 0000000..e2155ad
--- /dev/null
+++ b/topics/backend/data-pipeline/examples/incremental-sync/pipeline.py
@@ -0,0 +1,180 @@
+"""增量数据同步管道：第三方接口（限流 + 不稳定）→ 校验 → 幂等写入 → 数据质量检查 → 告警。
+
+只用标准库（sqlite3 代替真实数据库）。`python pipeline.py` 直接运行演示，test_pipeline.py 是测试。
+要点：
+- 水位线（watermark）：记录已同步到的 (updated_at, id)，下次从这里继续，可断点续跑
+- 限流：令牌桶控制调用频率，遇到 429 按 Retry-After 等待
+- 重试：只重试可重试的错误，指数退避 + 抖动，有次数上限
+- 幂等：按主键 upsert，并且只接受更新的版本，重复跑结果一样
+- 坏数据进入死信表，不阻塞整批；质量检查不通过就告警并且不推进水位线
+"""
+from __future__ import annotations
+
+import random
+import sqlite3
+import time
+from dataclasses import dataclass, field
+from typing import Callable, Iterable
+
+
+class RetryableError(Exception):
+    """网络超时、5xx、429 等，可以重试。"""
+
+    def __init__(self, msg: str, retry_after: float | None = None):
+        super().__init__(msg)
+        self.retry_after = retry_after
+
+
+class FatalError(Exception):
+    """401、参数错误等，重试没有意义。"""
+
+
+class QualityError(Exception):
+    """数据质量检查不通过。"""
+
+
+@dataclass
+class TokenBucket:
+    rate: float                     # 每秒补充的令牌数
+    capacity: float
+    now: Callable[[], float] = time.monotonic
+    sleep: Callable[[float], None] = time.sleep
+    tokens: float = field(init=False)
+    last: float = field(init=False)
+
+    def __post_init__(self) -> None:
+        self.tokens = self.capacity
+        self.last = self.now()
+
+    def acquire(self) -> None:
+        while True:
+            t = self.now()
+            self.tokens = min(self.capacity, self.tokens + (t - self.last) * self.rate)
+            self.last = t
+            if self.tokens >= 1:
+                self.tokens -= 1
+                return
+            self.sleep((1 - self.tokens) / self.rate)
+
+
+def with_retry(fn: Callable[[], list[dict]], *, attempts: int = 5, base: float = 0.2, cap: float = 5.0,
+               sleep: Callable[[float], None] = time.sleep, rng: random.Random | None = None) -> list[dict]:
+    rng = rng or random.Random()
+    for i in range(attempts):
+        try:
+            return fn()
+        except RetryableError as e:
+            if i == attempts - 1:
+                raise
+            delay = e.retry_after if e.retry_after is not None else rng.uniform(0, min(cap, base * 2 ** i))  # 全抖动
+            sleep(delay)
+    raise AssertionError("unreachable")
+
+
+SCHEMA = """
+CREATE TABLE IF NOT EXISTS orders (id TEXT PRIMARY KEY, amount_cents INTEGER NOT NULL, status TEXT NOT NULL, updated_at TEXT NOT NULL);
+CREATE TABLE IF NOT EXISTS dead_letters (source_id TEXT, reason TEXT, payload TEXT, at TEXT DEFAULT CURRENT_TIMESTAMP);
+CREATE TABLE IF NOT EXISTS watermarks (job TEXT PRIMARY KEY, updated_at TEXT NOT NULL, last_id TEXT NOT NULL);
+"""
+VALID_STATUS = {"paid", "refunded", "cancelled"}
+
+
+def validate(row: dict) -> str | None:
+    if not row.get("id"):
+        return "缺少 id"
+    if not isinstance(row.get("amount_cents"), int) or row["amount_cents"] < 0:
+        return "金额非法"
+    if row.get("status") not in VALID_STATUS:
+        return f"未知状态 {row.get('status')}"
+    if not row.get("updated_at"):
+        return "缺少 updated_at"
+    return None
+
+
+@dataclass
+class RunResult:
+    fetched: int = 0
+    written: int = 0
+    dead: int = 0
+    alerts: list[str] = field(default_factory=list)
+    committed: bool = False
+
+
+def run_sync(conn: sqlite3.Connection, fetch_page: Callable[[str, str, int], list[dict]], *,
+             job: str = "orders", page_size: int = 100, bucket: TokenBucket | None = None,
+             max_dead_ratio: float = 0.05, alert: Callable[[str], None] = print,
+             sleep: Callable[[float], None] = time.sleep) -> RunResult:
+    """从水位线开始分页拉取。fetch_page(after_updated_at, after_id, limit) 返回按 (updated_at, id) 升序的数据。"""
+    conn.executescript(SCHEMA)
+    wm = conn.execute("SELECT updated_at, last_id FROM watermarks WHERE job = ?", (job,)).fetchone()
+    after_ts, after_id = wm if wm else ("", "")
+    res = RunResult()
+    try:
+        with conn:                                   # 整次运行一个事务：质量检查不过就整体回滚，水位线不前进
+            while True:
+                if bucket:
+                    bucket.acquire()
+                page = with_retry(lambda: fetch_page(after_ts, after_id, page_size), sleep=sleep)
+                if not page:
+                    break
+                for row in page:
+                    res.fetched += 1
+                    reason = validate(row)
+                    if reason:
+                        res.dead += 1
+                        conn.execute("INSERT INTO dead_letters (source_id, reason, payload) VALUES (?, ?, ?)",
+                                     (str(row.get("id")), reason, repr(row)))
+                        continue
+                    # 幂等 upsert：只有来源版本更新时才覆盖，乱序或重复的数据不会把新值改回旧值
+                    conn.execute(
+                        """INSERT INTO orders (id, amount_cents, status, updated_at) VALUES (:id, :amount_cents, :status, :updated_at)
+                           ON CONFLICT(id) DO UPDATE SET amount_cents = excluded.amount_cents, status = excluded.status,
+                               updated_at = excluded.updated_at
+                           WHERE excluded.updated_at > orders.updated_at""", row)
+                    res.written += 1
+                last = page[-1]
+                after_ts, after_id = last.get("updated_at") or after_ts, str(last.get("id") or after_id)
+                if len(page) < page_size:
+                    break
+            # 数据质量检查
+            if res.fetched and res.dead / res.fetched > max_dead_ratio:
+                raise QualityError(f"坏数据比例 {res.dead}/{res.fetched} 超过阈值 {max_dead_ratio:.0%}")
+            conn.execute("INSERT INTO watermarks (job, updated_at, last_id) VALUES (?, ?, ?) "
+                         "ON CONFLICT(job) DO UPDATE SET updated_at = excluded.updated_at, last_id = excluded.last_id",
+                         (job, after_ts, after_id))
+            res.committed = True
+    except QualityError as e:
+        res.alerts.append(str(e))
+        alert(f"[告警] 任务 {job}：{e}，本次结果已回滚，水位线未推进")
+    return res
+
+
+
+class FakeSource:
+    """模拟第三方接口：按 (updated_at, id) 分页；可以注入 429 / 超时。"""
+
+    def __init__(self, rows: list[dict], failures: Iterable[Exception] = ()):
+        self.rows = sorted(rows, key=lambda r: (r["updated_at"], r["id"]))
+        self.failures = list(failures)
+        self.calls = 0
+
+    def __call__(self, after_ts: str, after_id: str, limit: int) -> list[dict]:
+        self.calls += 1
+        if self.failures:
+            raise self.failures.pop(0)
+        out = [r for r in self.rows if (r["updated_at"], r["id"]) > (after_ts, after_id)]
+        return [dict(r) for r in out[:limit]]
+
+
+if __name__ == "__main__":
+    rows = [{"id": f"o{i:03d}", "amount_cents": 100 * i, "status": "paid", "updated_at": f"2026-10-23T10:{i // 60:02d}:{i % 60:02d}"} for i in range(250)]
+    rows.append({"id": "bad1", "amount_cents": -5, "status": "paid", "updated_at": "2026-10-23T11:00:00"})
+    src = FakeSource(rows, failures=[RetryableError("429 Too Many Requests", retry_after=0.01), RetryableError("timeout")])
+    conn = sqlite3.connect(":memory:")
+    r1 = run_sync(conn, src, page_size=100, bucket=TokenBucket(rate=50, capacity=5), sleep=lambda s: None)
+    print(f"第一次：拉取 {r1.fetched}，写入 {r1.written}，死信 {r1.dead}，接口调用 {src.calls} 次（含 2 次失败重试）")
+    r2 = run_sync(conn, src, page_size=100, sleep=lambda s: None)
+    print(f"第二次（没有新数据）：拉取 {r2.fetched}")
+    total = conn.execute("SELECT COUNT(*) FROM orders").fetchone()[0]
+    print(f"orders 表 {total} 行，水位线 {conn.execute('SELECT updated_at, last_id FROM watermarks').fetchone()}")
+    assert (r1.written, r1.dead, r2.fetched, total) == (250, 1, 0, 250)
diff --git a/topics/backend/data-pipeline/examples/incremental-sync/test_pipeline.py b/topics/backend/data-pipeline/examples/incremental-sync/test_pipeline.py
new file mode 100644
index 0000000..269593a
--- /dev/null
+++ b/topics/backend/data-pipeline/examples/incremental-sync/test_pipeline.py
@@ -0,0 +1,80 @@
+import sqlite3
+
+import pytest
+
+from pipeline import FakeSource, FatalError, RetryableError, TokenBucket, run_sync, with_retry
+
+
+def make_rows(n, start=0, ts_prefix="2026-10-23T10:00:", status="paid"):
+    return [{"id": f"o{i:03d}", "amount_cents": i, "status": status, "updated_at": f"{ts_prefix}{i:02d}"} for i in range(start, start + n)]
+
+
+def test_resume_from_watermark_and_idempotent():
+    conn = sqlite3.connect(":memory:")
+    src = FakeSource(make_rows(30))
+    assert run_sync(conn, src, page_size=10, sleep=lambda s: None).written == 30
+    again = run_sync(conn, src, page_size=10, sleep=lambda s: None)
+    assert again.fetched == 0                                       # 水位线生效，不重复拉取
+    src.rows += make_rows(5, start=30)
+    assert run_sync(conn, src, page_size=10, sleep=lambda s: None).written == 5
+    assert conn.execute("SELECT COUNT(*) FROM orders").fetchone()[0] == 35
+
+
+def test_older_version_does_not_overwrite_newer():
+    conn = sqlite3.connect(":memory:")
+    run_sync(conn, FakeSource([{"id": "a", "amount_cents": 1, "status": "refunded", "updated_at": "2026-10-23T12:00:00"}]), job="j1")
+    # 另一个来源（或重放）送来旧版本
+    run_sync(conn, FakeSource([{"id": "a", "amount_cents": 1, "status": "paid", "updated_at": "2026-10-23T09:00:00"}]), job="j2")
+    assert conn.execute("SELECT status FROM orders WHERE id='a'").fetchone()[0] == "refunded"
+
+
+def test_quality_gate_rolls_back_and_alerts():
+    conn = sqlite3.connect(":memory:")
+    rows = make_rows(10) + make_rows(3, start=10, status="???")    # 3/13 坏数据，超过 5%
+    alerts = []
+    r = run_sync(conn, FakeSource(rows), max_dead_ratio=0.05, alert=alerts.append)
+    assert not r.committed and alerts
+    assert conn.execute("SELECT COUNT(*) FROM orders").fetchone()[0] == 0           # 整批回滚
+    assert conn.execute("SELECT COUNT(*) FROM watermarks").fetchone()[0] == 0       # 水位线没动
+
+
+def test_retry_only_retryable_and_respects_retry_after():
+    sleeps = []
+    calls = iter([RetryableError("429", retry_after=2.0), RetryableError("timeout"), ["ok"]])
+
+    def fn():
+        v = next(calls)
+        if isinstance(v, Exception):
+            raise v
+        return v
+
+    assert with_retry(fn, sleep=sleeps.append) == ["ok"]
+    assert sleeps[0] == 2.0 and 0 <= sleeps[1] <= 0.4
+
+    def fatal():
+        raise FatalError("401")
+
+    with pytest.raises(FatalError):
+        with_retry(fatal, sleep=sleeps.append)
+
+
+def test_retry_gives_up():
+    def always():
+        raise RetryableError("503")
+
+    with pytest.raises(RetryableError):
+        with_retry(always, attempts=3, sleep=lambda s: None)
+
+
+def test_token_bucket_limits_rate():
+    clock = [0.0]
+    slept = []
+
+    def sleep(s):
+        slept.append(s)
+        clock[0] += s
+
+    b = TokenBucket(rate=2, capacity=2, now=lambda: clock[0], sleep=sleep)
+    for _ in range(6):
+        b.acquire()
+    assert clock[0] == pytest.approx(2.0)            # 2 个初始令牌 + 4 个以每秒 2 个的速度补充 = 2 秒
diff --git a/topics/backend/data-pipeline/index.md b/topics/backend/data-pipeline/index.md
index bfb4993..bcf2473 100644
--- a/topics/backend/data-pipeline/index.md
+++ b/topics/backend/data-pipeline/index.md
@@ -1,15 +1,35 @@
 ---
 id: backend.data-pipeline.index
-title: 数据管道与第三方接口对接
+title: 数据管道基础
 type: index
 domain: backend
 stack: [data-pipeline]
-tags: []
+tags: [ETL, 增量同步, 幂等, 数据质量]
 week: [14]
 ---
 
-# 数据管道与第三方接口对接
+# 数据管道基础
 
-批处理与流处理、失败重试、幂等、数据质量告警、第三方 API 限流与字段变化。
+批处理与流处理、增量同步、限流重试、幂等、数据质量和告警。BOSS直聘 2026-10-23 的调研中部分 AI 应用岗位要求数据处理能力：知识库文档采集、第三方数据对接、报表数据准备都离不开它。
 
-> 本专题建设中：课程、企业场景、面试题和示例代码会陆续补充。
+## 学习路线
+
+| 顺序 | 课程 | 级别 | 预计 |
+| --- | --- | --- | --- |
+| 1 | [数据管道基础：批处理与流处理、增量同步、限流重试、幂等、数据质量和告警](01-batch-stream-reliability.md) | 进阶 | 3 小时 |
+
+## 企业场景
+
+- [第三方平台订单同步：限流、断点续跑、迟到数据、对账和告警](scenarios/third-party-sync.md)
+
+## 面试题
+
+- [数据管道面试题](interview.md)
+
+## 示例代码
+
+- [examples/incremental-sync](examples/incremental-sync/)：水位线、令牌桶、指数退避 + 抖动重试、幂等 upsert、死信、质量检查回滚（只用标准库，6 个测试通过）
+
+## 在 16 周计划中
+
+- [第 14 周：后端扩展与部署](../../../plan/week14-backend-extend-deploy.md)
diff --git a/topics/backend/data-pipeline/interview.md b/topics/backend/data-pipeline/interview.md
new file mode 100644
index 0000000..8c712b6
--- /dev/null
+++ b/topics/backend/data-pipeline/interview.md
@@ -0,0 +1,67 @@
+---
+id: backend.data-pipeline.interview
+title: 数据管道面试题
+type: interview
+domain: backend
+stack: [data-pipeline]
+tags: [数据同步, 幂等, 重试, 数据质量, 面试]
+level: intermediate
+related: [backend.data-pipeline.reliability]
+week: [14, 16]
+---
+
+# 数据管道面试题
+
+## 基础
+
+### 【基础】批处理和流处理的区别？
+
+**参考答案**：批处理按时间间隔处理一批数据，延迟高、实现简单，适合报表和对账；流处理事件到达即处理，延迟低，适合实时同步和告警。常结合使用：流处理保证实时，批处理兜底对账。
+
+**追问**：知识库文档更新后同步到向量库用哪种？→ 一般用事件驱动（流式）保证及时，再加每日对账修复遗漏。
+
+### 【基础】什么是幂等？同步任务怎么做到幂等？
+
+**参考答案**：同一操作执行多次与执行一次结果相同。做法：按业务主键 upsert；写入时比较版本号或更新时间；用唯一约束防止重复；外部副作用（发通知）记录已处理的 id。
+
+**追问**：为什么还要比较版本？→ 重试和乱序可能让旧数据后到，直接覆盖会把新状态改回旧状态。
+
+### 【基础】哪些错误应该重试？
+
+**参考答案**：临时性错误：超时、连接失败、5xx、429；不应该重试：认证失败、权限不足、参数错误等 4xx。重试要有上限、指数退避和抖动，429 遵守 `Retry-After`。
+
+**追问**：为什么要加抖动？→ 避免大量客户端在同一时刻重试造成“惊群”，把恢复中的服务再次打垮。
+
+## 进阶
+
+### 【进阶】增量同步的水位线怎么设计？
+
+**参考答案**：使用 (更新时间, 主键) 组合，按这个顺序分页拉取；本批数据成功落库后在同一事务中推进水位线；处理迟到数据时回拉一个重叠窗口，依赖幂等写入去重；定期对账兜底。
+
+**追问**：来源没有更新时间字段怎么办？→ 使用来源的自增 id / 变更日志 / CDC；都没有就只能定期全量比对（按哈希或汇总比较）。
+
+### 【进阶】怎么保证数据质量？
+
+**参考答案**：字段级校验（类型、必填、枚举、范围），坏数据进死信并保留原文；批次级检查（数量波动、空值率、唯一性）；与源头对账（数量、金额汇总）；异常时停止写入并告警；为质量规则写测试。
+
+**追问**：校验失败的数据怎么处理？→ 进入死信表，修复规则或上游后重新处理；比例超阈值整批回滚，避免污染下游。
+
+### 【进阶】多个工作进程调用同一个限流的第三方接口，怎么控制？
+
+**参考答案**：使用集中式限流：Redis 令牌桶 / 滑动窗口（Lua 脚本保证原子），按配额维度（平台 + 应用 / 商家）计数；拿不到令牌就等待或重新入队；监控 429 比例动态调整。
+
+**追问**：令牌桶和漏桶的区别？→ 令牌桶允许一定突发（桶内有积累的令牌）；漏桶以固定速率输出，更平滑。
+
+## 场景题
+
+### 【场景题】同步脚本中途挂掉，重跑后出现重复数据和漏数据，你怎么改？
+
+**参考答案**：引入水位线（落库成功后才推进）、幂等 upsert + 版本比较、重叠窗口处理迟到数据、单商家任务互斥、坏数据死信、批次质量检查和对账告警。详见[第三方订单同步场景](scenarios/third-party-sync.md)和[示例代码](examples/incremental-sync/)。
+
+**追问**：怎么验证修好了？→ 写测试：重复执行结果不变、中途失败后续跑不丢数据、旧版本不覆盖新版本、质量检查不通过时回滚。
+
+### 【场景题】RAG 知识库需要每天从多个来源（网盘、Wiki、工单系统）采集文档，你怎么设计管道？
+
+**参考答案**：每个来源一个连接器，按增量（修改时间 / 变更 API）拉取；统一处理流程：解析 → 清洗 → 分块 → 嵌入 → 写入向量库 / ES（按文档 id + 分块序号幂等）；删除和权限变更同步；嵌入接口限流和重试；记录每个文档的处理状态和版本；失败进死信可重试；质量检查（解析失败率、空文档、重复率）；对账确认来源和索引一致。
+
+**追问**：嵌入模型升级怎么办？→ 新建索引用新模型全量重建，评估通过后切换别名，旧索引保留回滚。
diff --git a/topics/backend/data-pipeline/scenarios/third-party-sync.md b/topics/backend/data-pipeline/scenarios/third-party-sync.md
new file mode 100644
index 0000000..87ca3ff
--- /dev/null
+++ b/topics/backend/data-pipeline/scenarios/third-party-sync.md
@@ -0,0 +1,81 @@
+---
+id: backend.data-pipeline.scenario.third-party-sync
+title: 第三方平台订单同步：限流、断点续跑、迟到数据、对账和告警
+type: scenario
+domain: backend
+stack: [data-pipeline, postgresql, redis, mq]
+tags: [第三方对接, 增量同步, 限流, 对账, 迟到数据, 告警]
+level: intermediate
+prerequisites: [backend.data-pipeline.reliability]
+est_hours: 2.5
+related: [backend.data-pipeline.interview, backend.redis.lock-ratelimit]
+week: [14]
+---
+
+# 第三方平台订单同步：限流、断点续跑、迟到数据、对账和告警
+
+## 业务背景与问题
+
+公司为商家提供 SaaS 经营分析和 AI 报表助手，需要把商家在多个电商平台的订单同步过来。现状是一个简单的定时脚本，问题频发：
+
+1. 平台接口限流，高峰期大量 429，脚本直接失败；
+2. 脚本中途挂掉后从头再跑，产生重复订单；
+3. 少量订单状态和平台对不上（退款了这边还是已支付）；
+4. 某次平台接口返回字段变更，脏数据写入了几个小时才被商家发现。
+
+## 约束
+
+- 每个平台的限流规则不同，而且按“商家授权的应用”计算配额。
+- 上千个商家，每个商家每 10 分钟同步一次。
+- 平台没有提供变更推送，或者推送不可靠。
+
+## 方案设计
+
+1. **任务拆分**：调度器每 10 分钟为每个“商家 × 平台”生成一个同步任务，投递到队列；多个工作进程消费；同一个商家同一时间只运行一个任务（分布式锁或队列分区）。
+2. **限流**：按“平台 + 应用”在 Redis 中做令牌桶，所有工作进程共享；429 按 `Retry-After` 退避。
+3. **断点续跑**：每个任务有自己的水位线（`updated_at + id`），本页数据落库后才推进；重叠窗口回拉 5 分钟处理迟到数据；upsert 带版本比较。
+4. **数据质量**：
+   - 字段级校验（类型、枚举、金额范围），坏数据进死信表并带原始报文；
+   - 批次级检查：坏数据比例、与上周同时段的数据量对比；
+   - 不通过就停止该平台的同步并告警，避免继续写入脏数据。
+5. **对账**：每天凌晨按商家拉取前一天的订单汇总（数量、金额、各状态数量）与本地比对，差异自动补拉并记录。
+6. **可观测**：每个任务记录拉取数、写入数、死信数、重试次数、耗时；面板按平台展示；连续失败、死信突增、对账差异超阈值时告警。
+
+## 关键代码
+
+基于 Redis 的共享令牌桶（Lua 保证原子性）：
+
+```lua
+-- KEYS[1]=桶的 key  ARGV: rate(每秒), capacity, now(毫秒)
+local b = redis.call('HMGET', KEYS[1], 'tokens', 'ts')
+local rate, cap, now = tonumber(ARGV[1]), tonumber(ARGV[2]), tonumber(ARGV[3])
+local tokens = tonumber(b[1]) or cap
+local ts = tonumber(b[2]) or now
+tokens = math.min(cap, tokens + (now - ts) / 1000 * rate)
+local ok = 0
+if tokens >= 1 then tokens = tokens - 1; ok = 1 end
+redis.call('HSET', KEYS[1], 'tokens', tokens, 'ts', now)
+redis.call('PEXPIRE', KEYS[1], math.ceil(cap / rate * 1000) + 1000)
+return ok
+```
+
+单进程版本的完整实现和测试：[examples/incremental-sync](../examples/incremental-sync/)。
+
+## 取舍
+
+| 选择 | 优点 | 缺点 |
+| --- | --- | --- |
+| **队列 + 工作进程 + 共享限流（本方案）** | 可水平扩展、互不影响 | 需要队列和 Redis |
+| 单个定时脚本串行同步 | 简单 | 商家多了跑不完，一个失败影响全部 |
+| 引入 Airflow 等编排平台 | 可视化、重跑方便 | 学习和运维成本，对小团队偏重 |
+
+## 易错点
+
+- 限流计数放在进程内存中，多个工作进程加起来超过配额。
+- 同一商家的两个任务并发运行，水位线互相覆盖。
+- 平台字段变更时没有校验，脏数据持续写入。
+- 只有“任务失败”告警，没有“数据量异常”告警：任务成功但拉回 0 条同样是问题。
+
+## 面试怎么讲
+
+> “我们对接了多个电商平台的订单。调度器为每个商家生成同步任务进队列，工作进程消费；限流用 Redis 令牌桶按平台和应用共享；每个任务有水位线，落库后才推进，回拉 5 分钟重叠窗口处理迟到数据，upsert 时比较更新时间防止旧数据覆盖。字段校验不过进死信，坏数据比例和数据量异常都会停止同步并告警。每天凌晨按汇总对账，差异自动补拉。”
diff --git a/topics/backend/elasticsearch/01-search-basics-hybrid.md b/topics/backend/elasticsearch/01-search-basics-hybrid.md
new file mode 100644
index 0000000..dea8a24
--- /dev/null
+++ b/topics/backend/elasticsearch/01-search-basics-hybrid.md
@@ -0,0 +1,96 @@
+---
+id: backend.elasticsearch.search-hybrid
+title: Elasticsearch：倒排索引、映射与中文分词、BM25、kNN 向量检索、混合检索和别名重建
+type: lesson
+domain: backend
+stack: [elasticsearch]
+tags: [Elasticsearch, 倒排索引, BM25, IK 分词, cjk, dense_vector, kNN, RRF, 混合检索, 别名, reindex]
+level: intermediate
+prerequisites: [ai.rag.hybrid-rerank]
+est_hours: 3
+related: [backend.elasticsearch.scenario.kb-search-sync, ai.rag.index, backend.elasticsearch.interview]
+week: [6, 14]
+order: 1
+---
+
+# Elasticsearch：倒排索引、映射与中文分词、BM25、kNN 向量检索、混合检索和别名重建
+
+> BOSS直聘 2026-10-23 的调研中，多个 RAG 岗位提到“ES 混合检索”。示例基于 Elasticsearch 9.5.1 + Python 客户端 9.5.1 实际运行通过。功能可用性与许可证有关（例如本地 basic 许可下服务端 RRF 返回 403），以官方文档为准：https://www.elastic.co/docs 。
+
+## 解释说明
+
+### 倒排索引和 BM25
+
+- **倒排索引**：词 → 包含这个词的文档列表。查询时先分词，再查倒排表，速度与文档总数关系不大。
+- **BM25**：ES 默认的相关性算法，考虑词频（有饱和上限）、逆文档频率（越稀有的词越重要）和文档长度。
+- **擅长**：精确词、专有名词、错误码、型号；**不擅长**：同义表达（“不想要了” vs “退货”）。
+
+### 映射和分词
+
+| 字段类型 | 用途 |
+| --- | --- |
+| `text` | 全文检索，写入时分词 |
+| `keyword` | 精确匹配、过滤、聚合、排序（租户、状态、标签） |
+| `dense_vector` | 向量，支持 kNN（HNSW） |
+
+中文分词：内置 `cjk` 分析器按二元切分，可用但粗糙；生产常用 **IK 分词插件**（`ik_max_word` 写入、`ik_smart` 查询）并维护业务词典和同义词，插件版本必须与 ES 版本匹配（以插件文档为准）。**映射一旦创建，已有字段的类型和分词器不能修改**，只能新建索引并重建数据。
+
+### 查询
+
+- `bool` 查询：`must`（参与算分）、`filter`（不算分、可缓存，用于租户 / 权限 / 状态）、`should`、`must_not`。
+- `multi_match` 多字段检索，`title^2` 提高标题权重。
+- `knn`：向量近似检索，`num_candidates` 越大越准越慢；`filter` 在检索过程中生效。
+
+### 混合检索
+
+关键词和向量各检索一次，用 **RRF（倒数排名融合）** 合并：只看名次，不需要把 BM25 分数和余弦相似度放在同一尺度上。ES 提供服务端的 `retriever.rrf`（许可证要求以官方文档为准），也可以在应用层融合（示例中的做法，任何许可都能用）。融合后通常再接**重排模型**（rerank）。
+
+### 别名和零停机重建
+
+业务只访问别名；修改映射时新建 `v2` 索引 → 全量导入 + 追增量 → 校验 → 原子切换别名 → 保留旧索引一段时间以便回滚。
+
+## 案例
+
+知识库问答：用户问“错误码 E1024 是什么”时关键词检索命中准确；问“商品不想要了怎么退”时依赖向量检索；两者 RRF 融合后再重排，取前几条给大模型。租户条件放在 `filter` 中，在两路检索中都生效。
+
+## 代码示例
+
+[examples/hybrid-search/es_hybrid.py](examples/hybrid-search/es_hybrid.py)（需要 `ES_URL`），实际输出：
+
+```text
+1. 关键词检索擅长精确词（错误码）：E1024 → ['d6']
+2. 租户过滤生效：t1 搜“退货” → ['d1']（没有 t2 的 d5）
+3. 向量检索（玩具嵌入）：['d6', 'd2', 'd1']
+4. RRF 混合检索：['d1', 'd2', 'd6']
+5. 服务端 RRF 不可用（AuthorizationException: ...current license is non-compli...），使用客户端 RRF
+6. 别名原子切换到 v2，查询不中断
+```
+
+（第 3 步用的是玩具嵌入，向量结果没有语义意义，只演示流程；真实项目换成嵌入模型。）
+
+```python
+def rrf(*rankings: list[str], k: int = 60) -> list[str]:
+    score: dict[str, float] = {}
+    for ranking in rankings:
+        for rank, doc_id in enumerate(ranking, start=1):
+            score[doc_id] = score.get(doc_id, 0.0) + 1.0 / (k + rank)
+    return sorted(score, key=lambda d: (-score[d], d))
+
+fused = rrf(bm25(q, tenant, 5), knn(q, tenant, 5))
+```
+
+## 易错点
+
+- 租户 / 权限条件写在 `must` 中（参与算分）或者在检索后才过滤（可能导致结果不足）。
+- 把需要精确匹配的字段设成 `text`。
+- 直接把 BM25 分数和向量相似度加权相加，尺度不同，权重很难调。
+- 修改映射后直接改原索引（做不到），或者没有别名导致切换时停机。
+- 用 ES 作为唯一的数据存储：ES 是检索引擎，数据源应在数据库中，ES 可以重建。
+- 本地关闭安全认证的配置被带到生产。
+
+## 面试常问
+
+- 倒排索引是什么？BM25 和 TF-IDF 的区别？
+- `text` 和 `keyword` 的区别？
+- 混合检索怎么做？为什么用 RRF？
+- 怎么在不停机的情况下修改映射？
diff --git a/topics/backend/elasticsearch/examples/hybrid-search/es_hybrid.py b/topics/backend/elasticsearch/examples/hybrid-search/es_hybrid.py
new file mode 100644
index 0000000..bda5ffd
--- /dev/null
+++ b/topics/backend/elasticsearch/examples/hybrid-search/es_hybrid.py
@@ -0,0 +1,136 @@
+# requires: elasticsearch
+"""Elasticsearch：中文分词、BM25 关键词检索、kNN 向量检索、RRF 混合检索、别名零停机重建索引。
+
+运行：ES_URL=http://127.0.0.1:9200 python es_hybrid.py
+本地测试用的单节点 ES 关闭了安全认证；生产环境必须开启认证和 TLS。没有设置环境变量时跳过。
+向量用一个“字符二元组哈希”的玩具嵌入代替真实模型，只为演示流程；索引名带随机后缀，结束后删除。
+"""
+import hashlib
+import math
+import os
+import sys
+import uuid
+
+from elasticsearch import Elasticsearch, helpers
+
+url = os.environ.get("ES_URL")
+if not url:
+    print("未设置 ES_URL，跳过")
+    sys.exit(0)
+
+es = Elasticsearch(url, request_timeout=30)
+suffix = uuid.uuid4().hex[:6]
+alias = f"kb-{suffix}"
+DIM = 64
+
+DOCS = [
+    {"id": "d1", "tenant": "t1", "title": "退货政策", "body": "签收后七天内可以申请无理由退货，商品需保持完好。"},
+    {"id": "d2", "tenant": "t1", "title": "退款到账时间", "body": "退款审核通过后，原路退回，一般一到三个工作日到账。"},
+    {"id": "d3", "tenant": "t1", "title": "发票开具", "body": "订单完成后可以在订单详情页申请电子发票。"},
+    {"id": "d4", "tenant": "t1", "title": "会员积分", "body": "每消费一元累计一个积分，积分可以抵扣运费。"},
+    {"id": "d5", "tenant": "t2", "title": "退货政策（其他租户）", "body": "本店商品不支持无理由退货。"},
+    {"id": "d6", "tenant": "t1", "title": "错误码 E1024", "body": "E1024 表示支付渠道超时，请稍后重试或更换支付方式。"},
+]
+
+
+def embed(text: str) -> list[float]:
+    """玩具嵌入：字符二元组哈希到 64 维再归一化。真实项目换成嵌入模型。"""
+    v = [0.0] * DIM
+    for a, b in zip(text, text[1:]):
+        h = int(hashlib.md5((a + b).encode()).hexdigest(), 16)
+        v[h % DIM] += 1.0
+    n = math.sqrt(sum(x * x for x in v)) or 1.0
+    return [x / n for x in v]
+
+
+def create_index(name: str) -> None:
+    es.indices.create(index=name, mappings={
+        "properties": {
+            "tenant": {"type": "keyword"},                       # 过滤用，不分词
+            "title": {"type": "text", "analyzer": "cjk"},        # 内置 cjk 分析器（中文二元切分）；生产常用 IK 插件，以插件文档为准
+            "body": {"type": "text", "analyzer": "cjk"},
+            "vec": {"type": "dense_vector", "dims": DIM, "index": True, "similarity": "cosine"},
+        }
+    })
+
+
+def load(index: str) -> None:
+    # 用业务 id 作为 _id：重复同步是覆盖而不是新增（幂等）
+    actions = ({"_index": index, "_id": d["id"], "_source": {**{k: v for k, v in d.items() if k != "id"}, "vec": embed(d["title"] + d["body"])}} for d in DOCS)
+    ok, errors = helpers.bulk(es, actions, refresh="wait_for")
+    assert ok == len(DOCS) and not errors
+
+
+def bm25(q: str, tenant: str, k: int = 3) -> list[str]:
+    r = es.search(index=alias, size=k, query={"bool": {
+        "must": {"multi_match": {"query": q, "fields": ["title^2", "body"]}},     # 标题权重更高
+        "filter": {"term": {"tenant": tenant}},                                    # 租户过滤放 filter：不算分、可缓存
+    }})
+    return [h["_id"] for h in r["hits"]["hits"]]
+
+
+def knn(q: str, tenant: str, k: int = 3) -> list[str]:
+    r = es.search(index=alias, size=k, knn={
+        "field": "vec", "query_vector": embed(q), "k": k, "num_candidates": 50,
+        "filter": {"term": {"tenant": tenant}},                                    # 过滤在向量检索过程中生效，而不是检索后再过滤
+    })
+    return [h["_id"] for h in r["hits"]["hits"]]
+
+
+def rrf(*rankings: list[str], k: int = 60) -> list[str]:
+    """倒数排名融合：只看名次，不看分数（BM25 分数和向量相似度不可直接比较）。"""
+    score: dict[str, float] = {}
+    for ranking in rankings:
+        for rank, doc_id in enumerate(ranking, start=1):
+            score[doc_id] = score.get(doc_id, 0.0) + 1.0 / (k + rank)
+    return sorted(score, key=lambda d: (-score[d], d))
+
+
+try:
+    v1 = f"{alias}-v1"
+    create_index(v1)
+    load(v1)
+    es.indices.put_alias(index=v1, name=alias)            # 业务只访问别名
+
+    exact = bm25("E1024", "t1")
+    assert exact[0] == "d6", exact
+    print(f"1. 关键词检索擅长精确词（错误码）：E1024 → {exact}")
+
+    kw = bm25("退货", "t1")
+    assert "d5" not in kw and kw[0] == "d1", kw
+    print(f"2. 租户过滤生效：t1 搜“退货” → {kw}（没有 t2 的 d5）")
+
+    sem = knn("商品不想要了怎么退", "t1")
+    print(f"3. 向量检索（玩具嵌入）：{sem}")
+    assert "d5" not in sem
+
+    fused = rrf(bm25("退货 退款多久到账", "t1", 5), knn("退货 退款多久到账", "t1", 5))
+    assert set(fused[:2]) == {"d1", "d2"}, fused
+    print(f"4. RRF 混合检索：{fused[:3]}")
+
+    # 服务端 RRF（retriever 语法）。许可证和可用版本以官方文档为准，不可用时退回客户端融合
+    try:
+        r = es.search(index=alias, size=3, retriever={"rrf": {"retrievers": [
+            {"standard": {"query": {"bool": {"must": {"match": {"body": "退款"}}, "filter": {"term": {"tenant": "t1"}}}}}},
+            {"knn": {"field": "vec", "query_vector": embed("退款到账"), "k": 3, "num_candidates": 50, "filter": {"term": {"tenant": "t1"}}}},
+        ]}})
+        print(f"5. 服务端 RRF：{[h['_id'] for h in r['hits']['hits']]}")
+    except Exception as e:  # noqa: BLE001  —— 例如 basic 许可下返回 403 授权错误
+        print(f"5. 服务端 RRF 不可用（{type(e).__name__}: {str(e)[:80]}），使用客户端 RRF")
+
+    # 别名切换：新建 v2（例如修改了映射），灌数据，原子切换别名，业务无感知
+    v2 = f"{alias}-v2"
+    create_index(v2)
+    load(v2)
+    es.indices.update_aliases(actions=[
+        {"remove": {"index": v1, "alias": alias}},
+        {"add": {"index": v2, "alias": alias}},
+    ])
+    assert list(es.indices.get_alias(name=alias).keys()) == [v2]
+    assert bm25("E1024", "t1")[0] == "d6"
+    print("6. 别名原子切换到 v2，查询不中断")
+    print("全部检查通过")
+finally:
+    # 生产集群通常禁止通配符删除（action.destructive_requires_name），这里逐个写明索引名
+    es.indices.delete(index=[f"{alias}-v1", f"{alias}-v2"], ignore_unavailable=True)
+    es.close()
diff --git a/topics/backend/elasticsearch/index.md b/topics/backend/elasticsearch/index.md
index 752dfd8..0731aa9 100644
--- a/topics/backend/elasticsearch/index.md
+++ b/topics/backend/elasticsearch/index.md
@@ -4,12 +4,38 @@ title: Elasticsearch
 type: index
 domain: backend
 stack: [elasticsearch]
-tags: []
+tags: [Elasticsearch, 全文检索, 混合检索, 向量检索]
 week: [6, 14]
 ---
 
 # Elasticsearch
 
-倒排索引、中文分词、BM25、与向量的混合检索、数据同步。
+倒排索引、中文分词、BM25 与 kNN 向量检索、混合检索和数据同步。BOSS直聘 2026-10-23 的调研中多个 RAG 岗位提到“ES 混合检索”，是 RAG 全链路的重要一环。
 
-> 本专题建设中：课程、企业场景、面试题和示例代码会陆续补充。
+## 学习路线
+
+| 顺序 | 课程 | 级别 | 预计 |
+| --- | --- | --- | --- |
+| 1 | [Elasticsearch：倒排索引、映射与中文分词、BM25、kNN 向量检索、混合检索和别名重建](01-search-basics-hybrid.md) | 进阶 | 3 小时 |
+
+## 企业场景
+
+- [知识库混合检索：数据库到 ES 的同步、权限过滤、分词词典和检索效果评估](scenarios/kb-search-sync.md)
+
+## 面试题
+
+- [Elasticsearch 面试题](interview.md)
+
+## 示例代码
+
+- [examples/hybrid-search](examples/hybrid-search/)：`cjk` 分词、租户过滤、BM25、kNN、应用层 RRF、别名切换（ES 9.5.1 单节点实际运行通过）
+
+## 在 16 周计划中
+
+- [第 6 周：RAG（二）](../../../plan/week06-rag-2.md)
+- [第 14 周：后端扩展与部署](../../../plan/week14-backend-extend-deploy.md)
+
+## 官方资料
+
+- https://www.elastic.co/docs
+- IK 分词插件：https://github.com/infinilabs/analysis-ik
diff --git a/topics/backend/elasticsearch/interview.md b/topics/backend/elasticsearch/interview.md
new file mode 100644
index 0000000..2ee9aa4
--- /dev/null
+++ b/topics/backend/elasticsearch/interview.md
@@ -0,0 +1,67 @@
+---
+id: backend.elasticsearch.interview
+title: Elasticsearch 面试题
+type: interview
+domain: backend
+stack: [elasticsearch]
+tags: [Elasticsearch, 倒排索引, 混合检索, 面试]
+level: intermediate
+related: [backend.elasticsearch.search-hybrid]
+week: [14, 16]
+---
+
+# Elasticsearch 面试题
+
+## 基础
+
+### 【基础】什么是倒排索引？
+
+**参考答案**：从“词”到“包含该词的文档列表（及位置、词频）”的映射。写入时对文本分词建立倒排表，查询时分词后查表求交并、算分，所以全文检索快。
+
+**追问**：为什么 ES 写入后不能立即搜到？→ 近实时：写入先进内存缓冲，`refresh`（默认约 1 秒）后生成新的段才可搜索；可以 `refresh=wait_for` 但会影响写入性能。
+
+### 【基础】`text` 和 `keyword` 的区别？
+
+**参考答案**：`text` 会分词，用于全文检索；`keyword` 不分词，用于精确匹配、过滤、排序和聚合。同一字段可以用多字段（`fields`）同时建两种。
+
+**追问**：对 `text` 字段做 `term` 查询会怎样？→ 往往查不到，因为存储的是分词后的词项，而 `term` 查询不分词。
+
+### 【基础】`must` 和 `filter` 的区别？
+
+**参考答案**：都要求匹配；`must` 参与相关性算分，`filter` 不算分并且可以被缓存，适合租户、权限、状态、时间范围等条件。
+
+**追问**：权限条件应该放哪里？→ `filter`，并且在向量检索中也要通过 `knn.filter` 传入。
+
+## 进阶
+
+### 【进阶】混合检索怎么做？为什么常用 RRF？
+
+**参考答案**：关键词（BM25）和向量（kNN）分别召回，再融合。BM25 分数无上限、与查询相关，向量相似度在固定区间，直接加权需要归一化且不稳定；RRF 只用名次 `1/(k+rank)` 求和，简单稳健。融合后再用重排模型精排。
+
+**追问**：服务端 RRF 和应用层 RRF 怎么选？→ 服务端少一次往返、写法简洁，但受版本和许可证限制；应用层灵活、可以融合来自不同系统（例如 ES + pgvector）的结果。
+
+### 【进阶】映射需要修改怎么办？
+
+**参考答案**：已有字段的类型和分词器不能修改。新建索引（新映射）→ `_reindex` 或从数据源重新导入 → 追平增量 → 校验 → 原子切换别名 → 保留旧索引以便回滚。
+
+**追问**：重建期间的增量写入怎么处理？→ 双写到新旧索引，或者重建完成后按时间戳 / 版本号补齐差异再切换。
+
+### 【进阶】中文分词怎么做？
+
+**参考答案**：常用 IK 插件：写入用细粒度（`ik_max_word`），查询用粗粒度（`ik_smart`）；维护业务词典和同义词；插件版本要与 ES 一致。内置 `cjk` 分析器按二元切分，可以作为没有插件时的方案。
+
+**追问**：词典更新后要做什么？→ 新写入的数据按新词典分词，已有数据需要重建索引才会生效。
+
+## 场景题
+
+### 【场景题】RAG 知识库对型号、错误码检索不准，怎么改？
+
+**参考答案**：引入关键词检索与向量检索混合（ES BM25 + kNN，RRF 融合 + 重排）；业务词典保证型号不被切碎；权限作为 filter；同步链路保证数据新鲜；用标注集评估召回率。详见[知识库混合检索场景](scenarios/kb-search-sync.md)。
+
+**追问**：怎么证明改进有效？→ 固定标注集，对比改动前后的召回率@K、MRR 以及最终回答的正确率。
+
+### 【场景题】数据库中的数据怎么可靠地同步到 ES？
+
+**参考答案**：常用方案：业务事务内写 outbox 表 → 投递到消息队列 → 幂等消费者写 ES（数据库主键作为 `_id`、带版本号防止旧数据覆盖新数据）；或者用 CDC（读取数据库日志，如 Debezium / MongoDB 变更流）。失败重试 + 死信 + 定期对账。
+
+**追问**：消息乱序导致旧版本覆盖新版本怎么办？→ 使用外部版本号（`version_type=external`）或写入前比较更新时间，只接受更新的版本。
diff --git a/topics/backend/elasticsearch/scenarios/kb-search-sync.md b/topics/backend/elasticsearch/scenarios/kb-search-sync.md
new file mode 100644
index 0000000..1554fba
--- /dev/null
+++ b/topics/backend/elasticsearch/scenarios/kb-search-sync.md
@@ -0,0 +1,85 @@
+---
+id: backend.elasticsearch.scenario.kb-search-sync
+title: 知识库混合检索：数据库到 ES 的同步、权限过滤、分词词典和检索效果评估
+type: scenario
+domain: backend
+stack: [elasticsearch, postgresql, mq, rag]
+tags: [混合检索, 数据同步, CDC, outbox, 权限过滤, IK 词典, 召回率]
+level: intermediate
+prerequisites: [backend.elasticsearch.search-hybrid, ai.rag.hybrid-rerank]
+est_hours: 2.5
+related: [backend.elasticsearch.interview, ai.rag.scenario.permission-aware-kb, backend.mq.reliability]
+week: [6, 14]
+---
+
+# 知识库混合检索：数据库到 ES 的同步、权限过滤、分词词典和检索效果评估
+
+## 业务背景与问题
+
+企业知识库 RAG 一开始只用了 pgvector 向量检索，问题：
+
+1. 用户搜产品型号、错误码、合同编号，向量检索经常找不到；
+2. 文档在后台修改后，问答仍引用旧内容；
+3. 偶尔检索到用户无权查看的部门文档；
+4. 每次调整分词或检索参数，不知道效果是变好还是变差。
+
+## 约束
+
+- 文档的主数据在 PostgreSQL（含权限）。
+- 修改后 1 分钟内可检索到。
+- 不能引入复杂的大数据组件。
+
+## 方案设计
+
+1. **检索层**：ES 索引同时保存分块的文本（IK 分词 + 业务词典：产品型号、内部术语）和向量；查询时 BM25 和 kNN 两路检索，应用层 RRF 融合，再用重排模型取前 5。
+2. **权限**：每个分块冗余 `tenant_id`、`dept_ids`、`visibility` 等 `keyword` 字段；查询时把当前用户的权限转换成 `filter`，两路检索都带上；返回前再按数据库做一次最终校验（防止同步延迟导致的权限变更未生效）。
+3. **同步**：
+   - 文档修改时在同一个数据库事务中写入 outbox 表；
+   - 后台任务读取 outbox 发到消息队列；
+   - 消费者重新分块、计算向量、按“文档 id + 分块序号”写入 ES（幂等），删除多余的旧分块；
+   - 失败重试，超过次数进入死信并告警；
+   - 每天对账：数据库与 ES 的文档版本号比对，修复差异。
+4. **变更映射**：通过别名 + 新索引重建，不停机。
+5. **评估**：维护一份标注集（问题 → 应该命中的文档），每次调整分词词典、权重、`num_candidates` 都计算召回率@K 和 MRR，比基线差就不上线。
+
+## 关键代码
+
+权限条件同时作用于两路检索：
+
+```python
+def acl_filter(user) -> list[dict]:
+    return [
+        {"term": {"tenant_id": user.tenant_id}},
+        {"bool": {"should": [
+            {"term": {"visibility": "public"}},
+            {"terms": {"dept_ids": user.dept_ids}},
+        ], "minimum_should_match": 1}},
+    ]
+
+kw = es.search(index="kb", size=20, query={"bool": {"must": {"match": {"text": q}}, "filter": acl_filter(user)}})
+vec = es.search(index="kb", size=20, knn={"field": "vec", "query_vector": embed(q), "k": 20,
+                                          "num_candidates": 200, "filter": acl_filter(user)})
+candidates = rrf(ids(kw), ids(vec))[:20]
+top5 = rerank(q, candidates)[:5]
+```
+
+完整可运行的基础示例：[examples/hybrid-search](../examples/hybrid-search/)；权限设计详见[权限感知知识库](../../../ai/rag/scenarios/permission-aware-kb.md)。
+
+## 取舍
+
+| 选择 | 优点 | 缺点 |
+| --- | --- | --- |
+| **ES 做混合检索（本方案）** | 关键词能力强、过滤和聚合成熟、向量也支持 | 多一套集群和同步链路 |
+| 只用 pgvector + PG 全文检索 | 组件少、强一致 | 中文分词和相关性调优能力弱一些 |
+| 专用向量库 + ES | 各自最强 | 两套检索系统，复杂度最高 |
+
+## 易错点
+
+- 只同步新增和修改，忘记处理删除和权限变更。
+- 消费者不是幂等的，重试后出现重复分块。
+- 业务词典更新后没有重建索引（已写入的数据仍按旧分词）。
+- 只凭几个例子判断效果，没有标注集和指标。
+
+## 面试怎么讲
+
+> “纯向量检索对型号、错误码这种精确词召回很差，我们引入 ES 做混合检索：IK 分词加业务词典，BM25 和 kNN 两路召回，应用层 RRF 融合后再重排。权限字段冗余到每个分块，作为 filter 同时作用于两路检索，返回前再校验一次。同步用 outbox + 消息队列，消费者按文档 id+分块号幂等写入，每天对账。每次调参都用标注集跑召回率@K 和 MRR，比基线差不上线。”
diff --git a/topics/backend/mongodb/01-document-model.md b/topics/backend/mongodb/01-document-model.md
new file mode 100644
index 0000000..97998ef
--- /dev/null
+++ b/topics/backend/mongodb/01-document-model.md
@@ -0,0 +1,105 @@
+---
+id: backend.mongodb.document-model
+title: MongoDB 基础：文档建模（嵌入还是引用）、索引与 ESR、聚合、事务和变更流
+type: lesson
+domain: backend
+stack: [mongodb]
+tags: [MongoDB, 文档模型, 嵌入, 引用, 复合索引, ESR, 聚合管道, 事务, 变更流, 副本集]
+level: basic
+prerequisites: [backend.postgresql.index]
+est_hours: 3
+related: [backend.mongodb.scenario.conversation-store, backend.elasticsearch.index, backend.mongodb.interview]
+week: [14]
+order: 1
+---
+
+# MongoDB 基础：文档建模（嵌入还是引用）、索引与 ESR、聚合、事务和变更流
+
+> BOSS直聘 2026-10-23 的调研中，有岗位把 MongoDB 列为要求或加分项（常见于存储对话记录、日志、灵活结构的业务数据）。定位：**了解并能用**，主库仍以 PostgreSQL / MySQL 为主。示例基于 MongoDB 8.2 + PyMongo 4.19 实际运行通过；以官方文档为准：https://www.mongodb.com/docs/manual/ 。
+
+## 解释说明
+
+### 和关系型数据库的对应
+
+| 关系型 | MongoDB |
+| --- | --- |
+| 数据库 / 表 / 行 | database / collection / document（BSON，类似 JSON） |
+| 表结构 | 默认无固定结构，可用 `$jsonSchema` 校验 |
+| JOIN | 优先“嵌入”，必要时用聚合的 `$lookup` |
+| 事务 | 单文档写入天然原子；多文档事务需要副本集或分片集群 |
+
+### 建模：嵌入还是引用
+
+- **嵌入**：总是一起读写、数量有上限的子数据（会话中的最近消息、订单中的商品快照）。一次读取拿到全部，单文档原子更新。
+- **引用**：数量无上限或会被独立访问、多处共享的数据（用户、商品主数据、海量消息）。
+- 单个文档上限 16MB，数组不能无限增长（“无限数组”是常见反模式）。
+
+### 索引
+
+- 单字段、复合、多键（数组）、TTL（自动过期）、唯一、文本索引等。
+- **复合索引的 ESR 规则**：等值条件字段在前（Equality），排序字段居中（Sort），范围条件字段在后（Range）。
+- 用 `explain()` 看是否 `IXSCAN`（走索引）、有没有 `COLLSCAN`（全表扫描）和 `SORT`（内存排序）。
+
+### 聚合管道
+
+`$match` → `$unwind` → `$group` → `$sort` → `$project` …… 像流水线一样处理；`$match` 尽量放在最前面以利用索引。
+
+### 副本集、事务、变更流
+
+- **副本集**：一主多从，自动故障转移；读写关注（`readConcern` / `writeConcern`）控制一致性。
+- **事务**：多文档 ACID，需要副本集；有性能成本，优先通过建模让一次写入只改一个文档。
+- **变更流（Change Streams）**：订阅集合的变更，常用于同步到 ES / 缓存 / 消息队列，支持 resume token 断点续传。
+
+## 案例
+
+AI 客服会话存储：会话文档嵌入最近消息（上限 200 条），结构随渠道不同而变化；按“租户 + 状态 + 时间倒序”查询待处理会话；转人工时同时改会话状态并创建工单（事务）；变更流把工单同步到搜索服务。
+
+## 代码示例
+
+[examples/conversations/mongo_demo.py](examples/conversations/mongo_demo.py)（需要 `MONGO_URL`，单节点副本集即可），实际输出：
+
+```text
+1. Schema 校验：非法 status 被拒绝
+2. 查询走索引且无需内存排序：['FETCH', 'IXSCAN', 'LIMIT']
+3. 聚合统计 6 组，例如 {'tenant': 't0', 'channel': 'app'} 共 50 个会话，平均 22.9 tokens
+4. 幂等 upsert：重复投递 3 次只有 1 条
+5. 事务：成功的提交生效，失败的整体回滚
+6. 变更流收到插入事件，并拿到 resume token 用于断点续传
+```
+
+ESR 索引和 explain：
+
+```python
+conv.create_index([("tenant_id", 1), ("status", 1), ("created_at", -1)])        # 等值、等值、排序
+plan = conv.find({"tenant_id": "t1", "status": "handoff"}).sort("created_at", -1).limit(10).explain()
+```
+
+幂等 upsert：
+
+```python
+events.create_index("event_id", unique=True)
+events.update_one({"event_id": "evt-001"}, {"$setOnInsert": {"payload": "退款申请"}}, upsert=True)
+```
+
+本地启动单节点副本集（无 Docker 时可用官方二进制包）：
+
+```bash
+mongod --dbpath ./data --replSet rs0 --bind_ip 127.0.0.1 --port 27017
+mongosh --eval 'rs.initiate()'      # 或用 PyMongo 执行 replSetInitiate
+```
+
+## 易错点
+
+- 把关系型的表结构原样搬过来，到处 `$lookup`，失去文档模型的优势。
+- 无限增长的嵌入数组。
+- 以为“无结构”就不需要校验：关键字段要用 Schema 校验和应用层校验。
+- 复合索引顺序不对，查询仍然内存排序。
+- 单机（非副本集）部署上使用事务或变更流，直接报错。
+- 用 `skip` 做深分页：越往后越慢，改用“基于上一页最后一个值”的范围查询。
+
+## 面试常问
+
+- 什么时候选 MongoDB，什么时候选 PostgreSQL？
+- 嵌入和引用怎么选？
+- 复合索引的设计原则？
+- MongoDB 支持事务吗？有什么限制？
diff --git a/topics/backend/mongodb/examples/conversations/mongo_demo.py b/topics/backend/mongodb/examples/conversations/mongo_demo.py
new file mode 100644
index 0000000..9cc02a5
--- /dev/null
+++ b/topics/backend/mongodb/examples/conversations/mongo_demo.py
@@ -0,0 +1,132 @@
+# requires: pymongo
+"""MongoDB：AI 客服会话存储——Schema 校验、复合索引与 explain、聚合统计、幂等 upsert、事务、变更流。
+
+运行：MONGO_URL="mongodb://127.0.0.1:27017/?directConnection=true" python mongo_demo.py
+需要 MongoDB 以副本集方式运行（事务和变更流都要求副本集；单节点副本集即可）。没有设置环境变量时跳过。
+使用一个随机名字的临时数据库，结束后删除。
+"""
+import os
+import sys
+import uuid
+from datetime import datetime, timedelta, timezone
+
+from pymongo import ASCENDING, DESCENDING, MongoClient
+from pymongo.errors import DuplicateKeyError, WriteError
+
+url = os.environ.get("MONGO_URL")
+if not url:
+    print("未设置 MONGO_URL，跳过")
+    sys.exit(0)
+
+client = MongoClient(url, serverSelectionTimeoutMS=3000)
+db = client[f"learn_{uuid.uuid4().hex[:8]}"]
+
+
+def plan_stages(plan: dict) -> set[str]:
+    """递归收集 explain 结果中的执行阶段名（IXSCAN / COLLSCAN / FETCH ...）。"""
+    out = {plan.get("stage", "")}
+    for key in ("inputStage", "queryPlan"):
+        if key in plan:
+            out |= plan_stages(plan[key])
+    for child in plan.get("inputStages", []):
+        out |= plan_stages(child)
+    return out
+
+
+try:
+    # 1. Schema 校验：文档模型灵活，但关键字段仍要约束
+    db.create_collection("conversations", validator={"$jsonSchema": {
+        "bsonType": "object",
+        "required": ["tenant_id", "user_id", "status", "created_at", "messages"],
+        "properties": {
+            "tenant_id": {"bsonType": "string"},
+            "status": {"enum": ["open", "handoff", "closed"]},
+            "messages": {"bsonType": "array", "maxItems": 200},   # 嵌入数组要设上限，防止文档无限增长（单文档上限 16MB）
+        },
+    }})
+    conv = db.conversations
+    try:
+        conv.insert_one({"tenant_id": "t1", "user_id": "u1", "status": "unknown", "created_at": datetime.now(timezone.utc), "messages": []})
+        raise AssertionError("非法 status 应该被拒绝")
+    except WriteError:
+        print("1. Schema 校验：非法 status 被拒绝")
+
+    # 2. 写入数据：消息嵌入在会话文档中（一起读、一起写，数量有上限）
+    now = datetime.now(timezone.utc)
+    docs = []
+    for i in range(300):
+        tenant = f"t{i % 3}"
+        docs.append({
+            "tenant_id": tenant, "user_id": f"u{i % 50}",
+            "status": ["open", "handoff", "closed"][i % 3],
+            "created_at": now - timedelta(minutes=i),
+            "channel": "web" if i % 2 else "app",
+            "messages": [{"role": "user", "text": f"问题{i}"}, {"role": "assistant", "text": f"回答{i}", "tokens": 20 + i % 7}],
+        })
+    conv.insert_many(docs)
+
+    # 3. 复合索引：遵循 ESR（等值 Equality → 排序 Sort → 范围 Range）
+    conv.create_index([("tenant_id", ASCENDING), ("status", ASCENDING), ("created_at", DESCENDING)], name="tenant_status_time")
+    query = {"tenant_id": "t1", "status": "handoff"}
+    plan = conv.find(query).sort("created_at", DESCENDING).limit(10).explain()["queryPlanner"]["winningPlan"]
+    stages = plan_stages(plan)
+    assert "IXSCAN" in stages and "COLLSCAN" not in stages and "SORT" not in stages, stages
+    print(f"2. 查询走索引且无需内存排序：{sorted(s for s in stages if s)}")
+
+    # 4. 聚合：每个租户每个渠道的会话数和平均 token
+    pipeline = [
+        {"$match": {"created_at": {"$gte": now - timedelta(hours=24)}}},
+        {"$unwind": "$messages"},
+        {"$match": {"messages.role": "assistant"}},
+        {"$group": {"_id": {"tenant": "$tenant_id", "channel": "$channel"}, "conversations": {"$sum": 1}, "avg_tokens": {"$avg": "$messages.tokens"}}},
+        {"$sort": {"_id.tenant": 1, "_id.channel": 1}},
+    ]
+    rows = list(conv.aggregate(pipeline))
+    assert sum(r["conversations"] for r in rows) == 300
+    print(f"3. 聚合统计 {len(rows)} 组，例如 {rows[0]['_id']} 共 {rows[0]['conversations']} 个会话，平均 {rows[0]['avg_tokens']:.1f} tokens")
+
+    # 5. 幂等写入：外部消息 id 做唯一索引 + upsert，重复投递不会产生重复数据
+    events = db.events
+    events.create_index("event_id", unique=True)
+    for _ in range(3):                                    # 同一条消息投递 3 次
+        events.update_one({"event_id": "evt-001"}, {"$setOnInsert": {"payload": "退款申请", "at": now}}, upsert=True)
+    assert events.count_documents({"event_id": "evt-001"}) == 1
+    try:
+        events.insert_one({"event_id": "evt-001"})
+        raise AssertionError("唯一索引应该拒绝")
+    except DuplicateKeyError:
+        pass
+    print("4. 幂等 upsert：重复投递 3 次只有 1 条")
+
+    # 6. 多文档事务：转人工时同时更新会话和创建工单，要么都成功要么都失败
+    tickets = db.tickets
+    target = conv.find_one({"status": "open"})
+    assert target is not None
+    with client.start_session() as s:
+        with s.start_transaction():
+            conv.update_one({"_id": target["_id"]}, {"$set": {"status": "handoff"}}, session=s)
+            tickets.insert_one({"conversation_id": target["_id"], "tenant_id": target["tenant_id"], "state": "new"}, session=s)
+    assert conv.find_one({"_id": target["_id"]})["status"] == "handoff" and tickets.count_documents({}) == 1
+    try:
+        with client.start_session() as s:
+            with s.start_transaction():
+                tickets.insert_one({"conversation_id": "x", "state": "new"}, session=s)
+                raise RuntimeError("模拟中途失败")
+    except RuntimeError:
+        pass
+    assert tickets.count_documents({}) == 1
+    print("5. 事务：成功的提交生效，失败的整体回滚")
+
+    # 7. 变更流：数据同步到搜索引擎 / 缓存的常用手段（可以用 resume token 断点续传）
+    with tickets.watch([{"$match": {"operationType": "insert"}}]) as stream:
+        tickets.insert_one({"conversation_id": "y", "tenant_id": "t2", "state": "new"})
+        change = stream.try_next()
+        while change is None:
+            change = stream.try_next()
+        token = stream.resume_token
+    assert change["fullDocument"]["conversation_id"] == "y" and token is not None
+    print("6. 变更流收到插入事件，并拿到 resume token 用于断点续传")
+    print("全部检查通过")
+finally:
+    client.drop_database(db.name)
+    client.close()
diff --git a/topics/backend/mongodb/index.md b/topics/backend/mongodb/index.md
index c7c736f..61be502 100644
--- a/topics/backend/mongodb/index.md
+++ b/topics/backend/mongodb/index.md
@@ -4,12 +4,37 @@ title: MongoDB
 type: index
 domain: backend
 stack: [mongodb]
-tags: []
+tags: [MongoDB, 文档数据库, NoSQL]
 week: [14]
 ---
 
 # MongoDB
 
-文档模型、查询与聚合、索引、何时选 MongoDB 而不是关系库。
+文档数据库的基础：建模、索引、聚合、事务、变更流。BOSS直聘 2026-10-23 的调研中有岗位提到，定位为“了解并能用”，主库仍以 PostgreSQL 为主。
 
-> 本专题建设中：课程、企业场景、面试题和示例代码会陆续补充。
+## 学习路线
+
+| 顺序 | 课程 | 级别 | 预计 |
+| --- | --- | --- | --- |
+| 1 | [MongoDB 基础：文档建模（嵌入还是引用）、索引与 ESR、聚合、事务和变更流](01-document-model.md) | 基础 | 3 小时 |
+
+## 企业场景
+
+- [AI 客服会话存储：海量对话记录的建模、查询、归档和同步](scenarios/conversation-store.md)
+
+## 面试题
+
+- [MongoDB 面试题](interview.md)
+
+## 示例代码
+
+- [examples/conversations](examples/conversations/)：Schema 校验、ESR 索引 + explain、聚合、幂等 upsert、事务、变更流（MongoDB 8.2 单节点副本集实际运行通过）
+
+## 在 16 周计划中
+
+- [第 14 周：后端扩展与部署](../../../plan/week14-backend-extend-deploy.md)
+
+## 官方资料
+
+- https://www.mongodb.com/docs/manual/
+- PyMongo：https://www.mongodb.com/docs/languages/python/pymongo-driver/current/
diff --git a/topics/backend/mongodb/interview.md b/topics/backend/mongodb/interview.md
new file mode 100644
index 0000000..86abfcc
--- /dev/null
+++ b/topics/backend/mongodb/interview.md
@@ -0,0 +1,67 @@
+---
+id: backend.mongodb.interview
+title: MongoDB 面试题
+type: interview
+domain: backend
+stack: [mongodb]
+tags: [MongoDB, 文档模型, 索引, 事务, 面试]
+level: basic
+related: [backend.mongodb.document-model]
+week: [14, 16]
+---
+
+# MongoDB 面试题
+
+## 基础
+
+### 【基础】MongoDB 和 MySQL / PostgreSQL 怎么选？
+
+**参考答案**：结构多变、以文档为单位读写、写入量大、可以接受反范式的数据（日志、对话、内容、配置）适合 MongoDB；强一致事务、复杂关联查询和报表（订单、账务）更适合关系型。很多系统组合使用。
+
+**追问**：PostgreSQL 的 JSONB 能不能代替？→ 中小规模、需要和关系数据一起做事务时完全可以；数据量大、结构差异大、需要水平分片时 MongoDB 更自然。
+
+### 【基础】嵌入和引用怎么选？
+
+**参考答案**：一起读写、数量有上限、不被单独访问的子数据嵌入；数量无上限、被多处共享或独立访问的数据引用。要考虑 16MB 单文档上限。
+
+**追问**：订单里的商品信息嵌入还是引用？→ 嵌入下单时的快照（价格、名称），同时保留商品 id 引用，因为订单要记录成交时的状态。
+
+### 【基础】MongoDB 支持事务吗？
+
+**参考答案**：单文档操作天然原子；4.0 起支持副本集上的多文档事务，4.2 起支持分片集群。事务有性能开销和时间限制，优先通过建模减少对多文档事务的依赖。
+
+**追问**：单机能用事务吗？→ 不能，需要副本集，开发环境可以启动单节点副本集。
+
+## 进阶
+
+### 【进阶】复合索引怎么设计？
+
+**参考答案**：按 ESR 规则：等值字段在前、排序字段中间、范围字段在后；考虑选择性和查询覆盖；用 `explain` 验证是否走索引、是否有内存排序。
+
+**追问**：为什么范围字段放在排序字段之前会有问题？→ 范围扫描后的结果在排序字段上不再有序，需要额外的内存排序。
+
+### 【进阶】深分页怎么优化？
+
+**参考答案**：避免大 `skip`（需要扫描并丢弃前面的文档），改用基于游标的分页：记录上一页最后一条的排序字段值（加 `_id` 打破平局），下一页查询“小于这个值”的数据。
+
+**追问**：这种分页的缺点？→ 不能直接跳到第 N 页，适合无限滚动和“下一页”场景。
+
+### 【进阶】变更流有什么用？怎么保证不丢事件？
+
+**参考答案**：订阅数据变更，用于同步到搜索引擎、缓存、消息队列、审计。消费者处理完后保存 resume token，重启时从 token 继续；消费端做幂等；注意 oplog 保留窗口，停太久 token 会失效，需要全量重建。
+
+**追问**：和 PostgreSQL 的什么机制类似？→ 逻辑复制 / CDC（如 Debezium 读取 WAL），以及 outbox 模式。
+
+## 场景题
+
+### 【场景题】AI 客服的对话记录怎么存储？
+
+**参考答案**：会话文档嵌入最近消息，完整消息单独集合并以 `conversation_id + seq` 唯一索引保证顺序和幂等；带 `tenant_id` 做隔离；按 ESR 建列表索引；变更流同步 ES 做检索；按保留期归档。详见[会话存储场景](scenarios/conversation-store.md)。
+
+**追问**：如何防止越权查询？→ 数据访问层统一注入 `tenant_id` 条件，并有测试覆盖“缺少租户条件时拒绝执行”。
+
+### 【场景题】线上一个查询突然变慢，你怎么排查？
+
+**参考答案**：开启慢查询分析（profiler 或日志），对慢查询执行 `explain("executionStats")` 看扫描的文档数与返回数之比、是否 COLLSCAN 或内存排序；检查是否新增了查询条件导致索引不匹配、数据分布变化、索引过多影响写入；修复后对比执行统计。
+
+**追问**：建索引会影响线上吗？→ 会占用资源，选择低峰期建立，大集合关注构建进度和副本同步延迟。
diff --git a/topics/backend/mongodb/scenarios/conversation-store.md b/topics/backend/mongodb/scenarios/conversation-store.md
new file mode 100644
index 0000000..22b6e93
--- /dev/null
+++ b/topics/backend/mongodb/scenarios/conversation-store.md
@@ -0,0 +1,78 @@
+---
+id: backend.mongodb.scenario.conversation-store
+title: AI 客服会话存储：海量对话记录的建模、查询、归档和同步
+type: scenario
+domain: backend
+stack: [mongodb, postgresql, elasticsearch]
+tags: [对话记录, 文档建模, TTL, 归档, 变更流, 多租户]
+level: intermediate
+prerequisites: [backend.mongodb.document-model]
+est_hours: 2
+related: [backend.mongodb.interview]
+week: [14]
+---
+
+# AI 客服会话存储：海量对话记录的建模、查询、归档和同步
+
+## 业务背景与问题
+
+AI 客服上线后，每天产生大量对话（用户消息、模型回复、工具调用记录、评分），原先存在 PostgreSQL 的一张 JSONB 大表中：
+
+1. 消息结构经常变化（新增工具调用、图片、引用来源），频繁改表；
+2. 坐席后台按租户查“待处理会话”越来越慢；
+3. 需要按关键词搜索历史对话；
+4. 数据只需要在线保留 180 天，之后归档。
+
+## 约束
+
+- 订单、退款等核心交易数据仍在 PostgreSQL，强一致。
+- 对话数据允许最终一致的检索。
+- 多租户隔离。
+
+## 方案设计
+
+1. **职责划分**：交易数据留在 PostgreSQL；对话记录迁移到 MongoDB（结构灵活、写入量大）；全文 / 语义检索交给 Elasticsearch。
+2. **建模**：
+   - `conversations`：会话元数据 + 最近 N 条消息嵌入（列表页和详情首屏一次读取）；
+   - `messages`：完整消息单独集合，按 `conversation_id + seq` 引用（避免无限数组）；
+   - 每个文档都带 `tenant_id`。
+3. **索引**：`{tenant_id, status, updated_at: -1}`（坐席列表，ESR）、`{conversation_id, seq}` 唯一（幂等写入和顺序读取）、`messages.created_at` 上的 TTL 索引或按月归档任务。
+4. **同步检索**：变更流 → 消费者 → 写入 ES；保存 resume token，断线后续传；消费者幂等（以消息 id 为 ES 文档 id）。
+5. **一致性**：转人工（改会话 + 建工单）用事务，或者把工单放在 PostgreSQL 并通过 outbox 模式保证一致。
+
+## 关键代码
+
+```python
+# 追加消息：唯一索引保证重复投递不重复写入；同时更新会话的最近消息（保留最后 20 条）
+messages.update_one(
+    {"conversation_id": cid, "seq": seq},
+    {"$setOnInsert": {"tenant_id": tenant, "role": role, "text": text, "created_at": now}},
+    upsert=True,
+)
+conversations.update_one(
+    {"_id": cid, "tenant_id": tenant},
+    {"$push": {"recent": {"$each": [{"seq": seq, "role": role, "text": text}], "$slice": -20}},
+     "$set": {"updated_at": now}},
+)
+```
+
+可运行的基础示例：[examples/conversations](../examples/conversations/)。
+
+## 取舍
+
+| 选择 | 优点 | 缺点 |
+| --- | --- | --- |
+| **MongoDB 存对话 + PG 存交易（本方案）** | 各取所长，结构灵活 | 多一种数据库要运维，跨库一致性要设计 |
+| 全部留在 PostgreSQL JSONB | 技术栈简单，事务方便 | 写入量大时表膨胀，结构变化仍需维护索引 |
+| 对话直接写 ES | 检索方便 | ES 不适合作为主存储（更新代价、一致性） |
+
+## 易错点
+
+- 消息全部嵌入会话文档，长会话超过 16MB。
+- 查询忘记带 `tenant_id`，越权读取其他租户数据。
+- 变更流消费者没有保存 resume token，重启后丢事件或全量重放。
+- 归档前没有确认合规要求（对话内容可能包含个人信息，需要脱敏或加密）。
+
+## 面试怎么讲
+
+> “对话记录结构多变、写入量大，我把它从 PG 的 JSONB 表迁到了 MongoDB，交易数据留在 PG。会话文档嵌入最近 20 条消息，完整消息单独集合用 conversation_id+seq 唯一索引，既避免无限数组又保证幂等。坐席列表按 ESR 原则建复合索引。检索通过变更流同步到 ES，保存 resume token 断点续传。180 天后归档。”
diff --git a/topics/backend/nodejs/01-express-nest-basics.md b/topics/backend/nodejs/01-express-nest-basics.md
new file mode 100644
index 0000000..59663c0
--- /dev/null
+++ b/topics/backend/nodejs/01-express-nest-basics.md
@@ -0,0 +1,92 @@
+---
+id: backend.nodejs.express-nest
+title: Node.js 后端（了解级）：事件循环、Express 5 中间件与错误处理、NestJS 的模块化思路、做 BFF
+type: lesson
+domain: backend
+stack: [nodejs]
+tags: [Node.js, 事件循环, Express 5, 中间件, 错误处理, NestJS, 依赖注入, BFF, SSE]
+level: basic
+prerequisites: [frontend.typescript.type-system]
+est_hours: 2.5
+related: [frontend.ssr.nextjs-app-router, backend.nodejs.scenario.ai-bff, backend.nodejs.interview]
+week: [7, 14]
+order: 1
+---
+
+# Node.js 后端（了解级）：事件循环、Express 5 中间件与错误处理、NestJS 的模块化思路、做 BFF
+
+> 定位：**了解并能写**。你的主线后端是 Go（基础）+ Python（AI / Agent）。BOSS直聘 2026-10-23 的调研中有岗位提到 Node.js / Next.js，前端出身的候选人被问到 Node 的概率很高。示例基于 Node 24 + Express 5 实际运行通过；NestJS 部分是示意代码，未编译。以官方文档为准：https://nodejs.org/docs 、https://expressjs.com/ 、https://docs.nestjs.com/ 。
+
+## 解释说明
+
+### 事件循环（面试必问）
+
+Node 单线程执行 JS，I/O 交给系统异步完成，回调放进队列。事件循环的主要阶段：timers（`setTimeout`）→ pending callbacks → poll（I/O）→ check（`setImmediate`）→ close callbacks；每个阶段之间先清空 `process.nextTick` 队列和 Promise 微任务队列。
+
+**结论**：适合 I/O 密集（BFF、网关、实时推送）；CPU 密集的计算会阻塞所有请求，要放到 `worker_threads` 或其他服务。
+
+### Express 5
+
+- **中间件**：`(req, res, next)` 按注册顺序执行，类似 Gin 的中间件链。
+- **错误处理中间件**：4 个参数 `(err, req, res, next)`，放在最后。
+- **Express 5 的变化**：async 处理函数抛出的错误 / 被拒绝的 Promise 会自动传给错误处理中间件（Express 4 需要手动 `next(err)`）；路由路径语法也有调整（以迁移指南为准）。
+
+### NestJS
+
+在 Express（或 Fastify）之上提供 Angular 风格的结构：模块（Module）、控制器（Controller）、服务（Provider）、**依赖注入**、管道（Pipe，做校验）、守卫（Guard，做鉴权）、拦截器（Interceptor）。适合中大型团队统一规范。对照：Go 中你手写依赖注入，FastAPI 用 `Depends`。
+
+### 为什么前端团队常用 Node 做 BFF
+
+同一语言、类型可以共享；聚合多个后端接口、裁剪字段；统一鉴权、限流；流式转发 AI 回答。
+
+## 案例
+
+AI 客服的 BFF：校验登录、限流、为每个请求生成 request id，调用 Python AI 服务并把流式回答以 SSE 转发给浏览器；客户端断开时中止上游。
+
+## 代码示例
+
+[examples/express-bff](examples/express-bff/)：`node --test app.test.ts` 4 个测试通过、`tsc` 通过，`node server.ts` 启动后 `curl -N` 观察到每 200ms 到达一条 SSE 事件。
+
+```ts
+// Express 5：async 路由中直接 throw，由统一错误处理中间件返回
+app.get("/orders/:id", async (req, res) => {
+  if (!/^[A-Z]\d{3,}$/.test(req.params.id)) throw new HttpError(400, "BAD_ORDER_ID", "订单号格式不正确");
+  const order = await deps.findOrder(req.params.id);
+  if (!order) throw new HttpError(404, "ORDER_NOT_FOUND", "订单不存在");
+  res.json(order);
+});
+
+app.use((err: unknown, _req: Request, res: Response, _next: NextFunction) => {
+  if (err instanceof HttpError) { res.status(err.status).json({ code: err.code, message: err.message }); return; }
+  res.status(500).json({ code: "INTERNAL", message: "服务器内部错误" });   // 不暴露内部细节
+});
+```
+
+NestJS 的写法（示意，未编译；需要装饰器相关的编译配置）：
+
+```ts
+@Controller("orders")
+export class OrdersController {
+  constructor(private readonly orders: OrdersService) {}          // 依赖注入
+
+  @Get(":id")
+  @UseGuards(AuthGuard)                                             // 鉴权
+  find(@Param("id") id: string) {
+    return this.orders.findOne(id);                                 // 抛出 NotFoundException 自动转成 404
+  }
+}
+```
+
+## 易错点
+
+- 在请求处理中做大量同步计算（大 JSON 序列化、加密、图片处理），阻塞所有请求。
+- Express 4 的项目里 async 错误没有 `next(err)`，请求挂起直到超时。
+- SSE 响应头发出后出错，还试图设置状态码。
+- 没有监听客户端断开，上游 AI 继续生成、浪费 token。
+- 未处理的 Promise 拒绝导致进程退出：要有全局监控和进程守护。
+
+## 面试常问
+
+- 讲讲 Node 的事件循环，`process.nextTick`、Promise、`setTimeout`、`setImmediate` 的顺序？
+- Express 中间件原理？错误处理怎么做？
+- Node 适合什么场景，不适合什么场景？
diff --git a/topics/backend/nodejs/examples/express-bff/app.test.ts b/topics/backend/nodejs/examples/express-bff/app.test.ts
new file mode 100644
index 0000000..f1f6879
--- /dev/null
+++ b/topics/backend/nodejs/examples/express-bff/app.test.ts
@@ -0,0 +1,53 @@
+// 不自动运行（需要 npm install 后执行 node --test app.test.ts）
+import assert from "node:assert/strict";
+import type { AddressInfo } from "node:net";
+import { after, before, test } from "node:test";
+import { createApp } from "./app.ts";
+
+let base = "";
+let close: () => void = () => {};
+
+before(async () => {
+  const app = createApp({
+    async *askLLM(q) {
+      if (q === "boom") throw new Error("upstream down");
+      yield "你好";
+      yield q;
+    },
+    findOrder: async (id) => (id === "A001" ? { id, status: "paid" } : null),
+  });
+  const server = app.listen(0);
+  await new Promise((r) => server.once("listening", r));
+  base = `http://127.0.0.1:${(server.address() as AddressInfo).port}`;
+  close = () => server.close();
+});
+after(() => close());
+
+test("查询订单：正常 / 格式错误 / 不存在", async () => {
+  assert.deepEqual(await (await fetch(`${base}/orders/A001`)).json(), { id: "A001", status: "paid" });
+  const bad = await fetch(`${base}/orders/xx`);
+  assert.equal(bad.status, 400);
+  assert.equal((await bad.json()).code, "BAD_ORDER_ID");
+  const missing = await fetch(`${base}/orders/A999`);
+  assert.equal(missing.status, 404);
+  assert.ok(missing.headers.get("x-request-id"));
+});
+
+test("请求体不是合法 JSON 时返回 400 而不是 500", async () => {
+  const r = await fetch(`${base}/chat`, { method: "POST", headers: { "content-type": "application/json" }, body: "{bad" });
+  assert.equal(r.status, 400);
+});
+
+test("流式对话：逐条 SSE 事件，最后是 done", async () => {
+  const r = await fetch(`${base}/chat`, { method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify({ question: "退货" }) });
+  assert.match(r.headers.get("content-type") ?? "", /^text\/event-stream/);   // express 会追加 charset
+  const text = await r.text();
+  assert.match(text, /data: \{"token":"你好"\}/);
+  assert.match(text, /data: \{"token":"退货"\}/);
+  assert.match(text, /event: done/);
+});
+
+test("上游失败：用 error 事件通知，而不是挂起连接", async () => {
+  const r = await fetch(`${base}/chat`, { method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify({ question: "boom" }) });
+  assert.match(await r.text(), /event: error/);
+});
diff --git a/topics/backend/nodejs/examples/express-bff/app.ts b/topics/backend/nodejs/examples/express-bff/app.ts
new file mode 100644
index 0000000..57fe84d
--- /dev/null
+++ b/topics/backend/nodejs/examples/express-bff/app.ts
@@ -0,0 +1,88 @@
+// 不自动运行：需要先 npm install（express 5）。Node 23.6+ 可以直接运行 .ts（类型擦除）
+// Express 5 BFF 示例：请求 id、输入校验、统一错误处理、async 路由（Express 5 会自动捕获 Promise 拒绝）、SSE 流式转发
+import express, { type NextFunction, type Request, type Response } from "express";
+import { randomUUID } from "node:crypto";
+
+export class HttpError extends Error {
+  status: number;
+  code: string;
+  constructor(status: number, code: string, message: string) {
+    super(message);
+    this.status = status;
+    this.code = code;
+  }
+}
+
+export type Deps = {
+  // 上游 AI 服务：返回逐段文本的异步迭代器；真实项目里是对 Python 服务的 fetch 流式调用
+  askLLM: (question: string, signal: AbortSignal) => AsyncIterable<string>;
+  findOrder: (id: string) => Promise<{ id: string; status: string } | null>;
+};
+
+export function createApp(deps: Deps) {
+  const app = express();
+  app.use(express.json({ limit: "100kb" }));                       // 限制请求体大小
+
+  app.use((req, res, next) => {                                      // 请求 id：贯穿日志和上游调用，便于排查
+    const id = req.get("x-request-id") ?? randomUUID();
+    res.locals.requestId = id;
+    res.set("x-request-id", id);
+    next();
+  });
+
+  app.get("/healthz", (_req, res) => {
+    res.json({ ok: true });
+  });
+
+  // Express 5：async 处理函数抛出的错误会自动交给错误处理中间件（Express 4 需要手动 next(err)）
+  app.get("/orders/:id", async (req, res) => {
+    const id = req.params.id;
+    if (!/^[A-Z]\d{3,}$/.test(id)) throw new HttpError(400, "BAD_ORDER_ID", "订单号格式不正确");
+    const order = await deps.findOrder(id);
+    if (!order) throw new HttpError(404, "ORDER_NOT_FOUND", "订单不存在");
+    res.json(order);
+  });
+
+  app.post("/chat", async (req, res) => {
+    const question: unknown = req.body?.question;
+    if (typeof question !== "string" || question.length === 0 || question.length > 500) {
+      throw new HttpError(400, "BAD_QUESTION", "question 必须是 1~500 字的字符串");
+    }
+    const controller = new AbortController();
+    res.on("close", () => controller.abort());                       // 客户端断开：通知上游停止生成
+    res.set({ "Content-Type": "text/event-stream", "Cache-Control": "no-cache", "X-Accel-Buffering": "no" });
+    res.flushHeaders();
+    try {
+      for await (const token of deps.askLLM(question, controller.signal)) {
+        if (controller.signal.aborted) break;
+        res.write(`data: ${JSON.stringify({ token })}\n\n`);
+      }
+      res.write("event: done\ndata: {}\n\n");
+    } catch (e) {
+      // 响应头已经发出，不能再改状态码：用一个 error 事件告诉前端
+      res.write(`event: error\ndata: ${JSON.stringify({ message: "生成失败，请重试" })}\n\n`);
+      console.error(res.locals.requestId, e);
+    } finally {
+      res.end();
+    }
+  });
+
+  app.use((_req, _res, next) => next(new HttpError(404, "NOT_FOUND", "接口不存在")));
+
+  // 统一错误处理：4 个参数。不把内部错误细节返回给客户端
+  app.use((err: unknown, _req: Request, res: Response, _next: NextFunction) => {
+    if (err instanceof HttpError) {
+      res.status(err.status).json({ code: err.code, message: err.message, requestId: res.locals.requestId });
+      return;
+    }
+    const status = (err as { status?: number }).status;
+    if (status === 400 || status === 413) {                          // express.json 的解析错误 / 请求体过大
+      res.status(status).json({ code: "BAD_REQUEST", message: "请求体无效", requestId: res.locals.requestId });
+      return;
+    }
+    console.error(res.locals.requestId, err);
+    res.status(500).json({ code: "INTERNAL", message: "服务器内部错误", requestId: res.locals.requestId });
+  });
+
+  return app;
+}
diff --git a/topics/backend/nodejs/examples/express-bff/package.json b/topics/backend/nodejs/examples/express-bff/package.json
new file mode 100644
index 0000000..c0f056b
--- /dev/null
+++ b/topics/backend/nodejs/examples/express-bff/package.json
@@ -0,0 +1,17 @@
+{
+  "name": "express-bff-demo",
+  "private": true,
+  "type": "module",
+  "scripts": {
+    "start": "node server.ts",
+    "test": "node --test app.test.ts"
+  },
+  "dependencies": {
+    "express": "^5.1.0"
+  },
+  "devDependencies": {
+    "@types/express": "^5.0.0",
+    "@types/node": "^24.0.0",
+    "typescript": "^5.9.0"
+  }
+}
diff --git a/topics/backend/nodejs/examples/express-bff/server.ts b/topics/backend/nodejs/examples/express-bff/server.ts
new file mode 100644
index 0000000..c195311
--- /dev/null
+++ b/topics/backend/nodejs/examples/express-bff/server.ts
@@ -0,0 +1,17 @@
+// 不自动运行：启动服务 node server.ts，然后 curl -N -X POST localhost:3000/chat -H 'content-type: application/json' -d '{"question":"退货"}'
+import { createApp } from "./app.ts";
+
+async function* fakeLLM(q: string, signal: AbortSignal) {
+  for (const t of ["关于", `“${q}”`, "：", "签收 7 天内", "可以申请退货。"]) {
+    if (signal.aborted) return;
+    await new Promise((r) => setTimeout(r, 200));
+    yield t;
+  }
+}
+
+const app = createApp({
+  askLLM: fakeLLM,
+  findOrder: async (id) => (id === "A001" ? { id, status: "paid" } : null),
+});
+const port = Number(process.env.PORT ?? 3000);
+app.listen(port, () => console.log(`listening on http://127.0.0.1:${port}`));
diff --git a/topics/backend/nodejs/examples/express-bff/tsconfig.json b/topics/backend/nodejs/examples/express-bff/tsconfig.json
new file mode 100644
index 0000000..ae27c4b
--- /dev/null
+++ b/topics/backend/nodejs/examples/express-bff/tsconfig.json
@@ -0,0 +1,14 @@
+{
+  "compilerOptions": {
+    "target": "ES2023",
+    "module": "NodeNext",
+    "moduleResolution": "NodeNext",
+    "strict": true,
+    "noEmit": true,
+    "skipLibCheck": true,
+    "allowImportingTsExtensions": true,
+    "erasableSyntaxOnly": true,
+    "verbatimModuleSyntax": true
+  },
+  "include": ["*.ts"]
+}
diff --git a/topics/backend/nodejs/index.md b/topics/backend/nodejs/index.md
index 6363077..82df83f 100644
--- a/topics/backend/nodejs/index.md
+++ b/topics/backend/nodejs/index.md
@@ -1,15 +1,36 @@
 ---
 id: backend.nodejs.index
-title: Node.js 后端（Express / NestJS）
+title: Node.js（了解级）
 type: index
 domain: backend
 stack: [nodejs]
-tags: []
+tags: [Node.js, Express, NestJS, BFF]
 week: [7, 14]
 ---
 
-# Node.js 后端（Express / NestJS）
+# Node.js（了解级）
 
-了解级：前端出身的后端选项，Express 中间件、NestJS 模块与依赖注入。
+事件循环、Express 5、NestJS 思路，以及为 AI 应用做 BFF。定位“了解并能写”：主线后端是 Go 基础 + Python AI。BOSS直聘 2026-10-23 的调研中有岗位提到 Node.js / Next.js。
 
-> 本专题建设中：课程、企业场景、面试题和示例代码会陆续补充。
+## 学习路线
+
+| 顺序 | 课程 | 级别 | 预计 |
+| --- | --- | --- | --- |
+| 1 | [Node.js 后端（了解级）：事件循环、Express 5 中间件与错误处理、NestJS 的模块化思路、做 BFF](01-express-nest-basics.md) | 基础 | 2.5 小时 |
+
+## 企业场景
+
+- [用 Node.js 给 AI 应用做 BFF：鉴权、限流、流式转发、超时和断开处理](scenarios/ai-bff.md)
+
+## 面试题
+
+- [Node.js 面试题](interview.md)
+
+## 示例代码
+
+- [examples/express-bff](examples/express-bff/)：Express 5 统一错误处理、request id、SSE 流式输出与断开处理（4 个测试通过，tsc 通过）
+
+## 在 16 周计划中
+
+- [第 7 周：前端工程化查漏](../../../plan/week07-frontend.md)
+- [第 14 周：后端扩展与部署](../../../plan/week14-backend-extend-deploy.md)
diff --git a/topics/backend/nodejs/interview.md b/topics/backend/nodejs/interview.md
new file mode 100644
index 0000000..8043ff1
--- /dev/null
+++ b/topics/backend/nodejs/interview.md
@@ -0,0 +1,67 @@
+---
+id: backend.nodejs.interview
+title: Node.js 面试题
+type: interview
+domain: backend
+stack: [nodejs]
+tags: [Node.js, 事件循环, Express, NestJS, 面试]
+level: basic
+related: [backend.nodejs.express-nest]
+week: [14, 16]
+---
+
+# Node.js 面试题
+
+## 基础
+
+### 【基础】Node 的事件循环是怎样的？
+
+**参考答案**：单线程执行 JS，I/O 由 libuv 异步完成。主要阶段：timers → pending callbacks → poll → check → close callbacks；每个回调执行完后先清空 `process.nextTick` 队列，再清空 Promise 微任务队列。
+
+**追问**：`setTimeout(fn, 0)` 和 `setImmediate` 谁先执行？→ 在主模块中顺序不确定（取决于进入事件循环时计时器是否到期）；在 I/O 回调中 `setImmediate` 一定先执行。
+
+### 【基础】Node 适合什么场景？
+
+**参考答案**：I/O 密集、高并发连接：BFF、API 网关、实时推送（WebSocket / SSE）、工具链。不适合长时间的 CPU 计算，否则阻塞事件循环；需要时用 `worker_threads` 或拆到其他服务。
+
+**追问**：怎么发现事件循环被阻塞？→ 监控事件循环延迟（`perf_hooks.monitorEventLoopDelay`），用 CPU profile 找热点。
+
+### 【基础】Express 中间件的执行原理？
+
+**参考答案**：按注册顺序组成链，每个中间件处理后调用 `next()` 进入下一个，不调用则请求在此结束；错误处理中间件有 4 个参数，`next(err)` 或 Express 5 中 async 函数抛错时跳转过去。
+
+**追问**：和 Koa 的洋葱模型有什么区别？→ Koa 的 `await next()` 可以在后续中间件执行完后继续执行后半段逻辑，天然适合计时和统一响应处理；Express 需要监听 `res` 的 `finish` 事件实现类似效果。
+
+## 进阶
+
+### 【进阶】NestJS 的依赖注入有什么好处？
+
+**参考答案**：类声明依赖，由容器创建和注入；便于替换实现（测试中注入假的服务）、统一生命周期管理、模块边界清晰；配合管道、守卫、拦截器形成统一规范。
+
+**追问**：和 FastAPI 的 `Depends` 比较？→ 思想相同，FastAPI 基于函数参数声明，Nest 基于类构造函数和装饰器元数据。
+
+### 【进阶】Node 服务如何处理未捕获的异常？
+
+**参考答案**：请求级错误用统一错误处理中间件；进程级监听 `uncaughtException` / `unhandledRejection` 记录日志后退出，由进程管理器（容器编排 / systemd / PM2）重启；不要在未知状态下继续运行。
+
+**追问**：为什么不能捕获后继续运行？→ 程序可能处于不一致的状态（连接、内存中的数据），继续运行可能产生更隐蔽的错误。
+
+### 【进阶】Node 如何利用多核？
+
+**参考答案**：多进程（`cluster` 或容器多副本 + 负载均衡）处理更多请求；`worker_threads` 处理 CPU 计算。服务要无状态，会话等放在 Redis。
+
+**追问**：WebSocket 多实例怎么广播？→ 通过 Redis Pub/Sub 或消息队列在实例之间转发，负载均衡开启会话保持或使用支持多实例的方案。
+
+## 场景题
+
+### 【场景题】给 AI 应用做 BFF，你会做哪些事？
+
+**参考答案**：统一鉴权、限流、request id；首屏并行聚合与降级；SSE 流式透传、断开时中止上游、超时控制、上游错误以事件形式通知；保持薄层。详见[AI BFF 场景](scenarios/ai-bff.md)。
+
+**追问**：怎么测试流式接口？→ 启动应用监听随机端口，用 `fetch` 读取响应，断言事件内容和顺序、错误事件和响应头，见[示例测试](examples/express-bff/app.test.ts)。
+
+### 【场景题】Node 服务内存持续上涨，怎么排查？
+
+**参考答案**：观察堆内存趋势；在不同时间点生成堆快照对比，找增长的对象和引用链；常见原因：全局缓存无上限、事件监听未移除、闭包持有大对象、定时器未清理。修复后压测验证。
+
+**追问**：线上怎么安全地拿堆快照？→ 在摘除流量的单个实例上进行，快照期间进程会暂停，注意快照可能包含敏感数据。
diff --git a/topics/backend/nodejs/scenarios/ai-bff.md b/topics/backend/nodejs/scenarios/ai-bff.md
new file mode 100644
index 0000000..633c976
--- /dev/null
+++ b/topics/backend/nodejs/scenarios/ai-bff.md
@@ -0,0 +1,84 @@
+---
+id: backend.nodejs.scenario.ai-bff
+title: 用 Node.js 给 AI 应用做 BFF：鉴权、限流、流式转发、超时和断开处理
+type: scenario
+domain: backend
+stack: [nodejs, redis, nginx, fastapi]
+tags: [BFF, SSE, 流式转发, 限流, 超时, request id]
+level: intermediate
+prerequisites: [backend.nodejs.express-nest]
+est_hours: 2
+related: [backend.nodejs.interview, frontend.ssr.scenario.help-center]
+week: [14]
+---
+
+# 用 Node.js 给 AI 应用做 BFF：鉴权、限流、流式转发、超时和断开处理
+
+## 业务背景与问题
+
+前端团队维护的 AI 助手页面直接调用 Python 的 Agent 服务，问题：
+
+1. Agent 服务要校验各业务线不同的登录态，越来越臃肿；
+2. 页面需要的数据来自 3 个服务，前端串行请求，首屏慢；
+3. 用户关掉页面后，Agent 仍在生成，浪费模型费用；
+4. 出问题时无法串起前端、BFF、Agent 三方日志。
+
+## 约束
+
+- 前端团队熟悉 TypeScript，希望自己维护这一层。
+- Agent 服务由 AI 团队维护，接口为 SSE。
+
+## 方案设计
+
+1. **BFF（Express 5 或 NestJS）**：统一鉴权（会话 / JWT）、按用户限流（Redis）、生成并透传 `x-request-id`。
+2. **聚合接口**：首屏所需的用户信息、会话列表、配额用 `Promise.allSettled` 并行请求，各自设超时，部分失败时降级返回。
+3. **流式转发**：`fetch` 上游时传入 `AbortSignal`，`res.on("close")` 时中止；首字节超时和总超时分别控制；上游出错时发送 SSE `error` 事件而不是挂起。
+4. **部署**：多实例无状态，Nginx 关闭对 SSE 路由的缓冲，设置较长的读超时。
+
+## 关键代码
+
+```ts
+app.post("/chat", async (req, res) => {
+  const user = await auth(req);                                   // 失败时 throw HttpError(401)
+  await rateLimit(`chat:${user.id}`);                             // 超限 throw HttpError(429)
+  const controller = new AbortController();
+  res.on("close", () => controller.abort());
+  const timeout = setTimeout(() => controller.abort(), 120_000);  // 总超时
+  const upstream = await fetch(`${AGENT_URL}/chat`, {
+    method: "POST",
+    headers: { "content-type": "application/json", "x-request-id": res.locals.requestId, "x-user-id": user.id },
+    body: JSON.stringify({ question: req.body.question }),
+    signal: controller.signal,
+  });
+  res.set({ "Content-Type": "text/event-stream", "Cache-Control": "no-cache", "X-Accel-Buffering": "no" });
+  try {
+    for await (const chunk of upstream.body!) res.write(chunk);  // 原样透传上游的 SSE 字节
+  } catch {
+    if (!controller.signal.aborted) res.write('event: error\ndata: {"message":"生成中断"}\n\n');
+  } finally {
+    clearTimeout(timeout);
+    res.end();
+  }
+});
+```
+
+可运行的基础版本（含测试）：[examples/express-bff](../examples/express-bff/)。
+
+## 取舍
+
+| 选择 | 优点 | 缺点 |
+| --- | --- | --- |
+| **Node BFF（本方案）** | 前端团队自治、类型共享、擅长 I/O | 多一层服务和一跳网络 |
+| Next.js 路由处理器做 BFF | 和页面同一个项目 | 受部署形态限制（Serverless 超时） |
+| 由 Agent 服务直接对外 | 链路短 | 鉴权、聚合逻辑侵入 AI 服务 |
+
+## 易错点
+
+- 把上游的流读完再返回（`await upstream.text()`），失去流式效果。
+- 没有向上游传播取消。
+- `Promise.all` 中一个接口失败导致整页失败：首屏聚合用 `allSettled` + 降级。
+- 在 BFF 里堆业务逻辑，最后变成“第二个后端”。
+
+## 面试怎么讲
+
+> “AI 助手页面我们加了一层 Node BFF：统一鉴权和按用户限流，生成 request id 透传到 Agent 服务；首屏数据并行聚合、部分失败降级；对话接口把 Agent 的 SSE 原样透传，用户断开时通过 AbortController 中止上游，节省模型费用；上游失败时发 error 事件。BFF 保持薄，只做协议适配和横切逻辑。”