> For AI agents: the complete documentation index is available at /llms.txt, the full documentation bundle is available at /llms-full.txt. # Lambda 写入矩阵 每个 Lambda 对每张 Neon 表的字段级写入职责清单,给开发者改 Lambda 写入逻辑前对照。**字段所有权是硬约束** — 违反会导致 last-write-wins 竞态或数据不一致。 ::: details 实施溯源 + 落地状态(2026-05-04) **设计文档**: - Pipeline 设计: `.claude/specs/2026-03-25-task-pipeline-write-flow.md` - Flow 3(AI-driven task decisions): `.claude/specs/2026-03-28-flow3-contact-analysis-design.md` **Schema 状态**(callytics-common `0.34.0` + migration `0014-0023`): - 主业务表 `contacts` / `calls` / `messages` / `leads` / `contact_timeline` / `tasks` / `staff`,辅助表 `rc-stores` / `store-config` / `user-stores` 均具备 `store_id` 列 - **`task_events` 仍只有 `account_id`,无 `store_id`** — audit 表暂保留 account-level isolation,store-level 终态后续补 - `contacts` 复合主键 = `(phone, store_id)`,`store_id NOT NULL`(migration 0014) - `staff.store_id` NOT NULL(migration 0015) - `site_id` → `account_id` rename(migration 0016) - 其余表 `store_id` 仍 nullable(`calls` / `leads` / `messages` / `contact_timeline` / `tasks`),writer 全写但保留 NULL 兼容历史行,各表均有 `idx_<table>_store_id WHERE store_id IS NOT NULL` partial index **Lambda writer 当前 store\_id 写入状态**(2026-05-04): - `transcribe-processor` → `calls.store_id` + `contact_timeline.store_id`(`call.status_changed`,Phase 1 起 dual writer) - `ai-analysis-processor` → `contacts/calls/contact_timeline.store_id` - `contacts-analyzer` → `contacts/tasks/contact_timeline.store_id` - `message-processor` → `messages/contacts/contact_timeline.store_id` - `lead-tracking`(独立 repo)→ `leads/contacts/tasks/contact_timeline.store_id` **`franchise_id` 来源**(读源代码核对): - `message-processor`:`parseFranchiseFromClientId(_client_id, _account_id)` 解析,**不是硬编码** - `ai-analysis-processor`:从 SQS 上游 metadata 传入 - `contacts-analyzer`:从 contact 行的 `franchiseId` 读取 - `lead-tracking`:StoresV2 leadEmail 映射 → `parseBrandFromFranchise()` - 当前所有 writer 都从外部数据派生 `franchise_id`,**不硬编码**(历史 P0 #632 fix:阻止"`_client_id` 整串含 site 后缀直接写入"的 bug)。 ## ::: ## 全局数据流 ```mermaid %%{init: {'theme': 'base', 'themeVariables': {'primaryColor': '#f5f5f5', 'lineColor': '#666'}}}%% flowchart TB subgraph entry["入口层"] LEAD["📧 Email Lead"] CALL["📞 通话结束"] SMS["💬 SMS/VM"] UI["👤 员工 UI\n(DEFERRED)"] end subgraph lt["Lead Processing"] LT["lead-tracking\n(独立 repo)\n非阻塞写入"] end subgraph p1["Per-Call Analysis"] AAP["ai-analysis-processor\n(Prompt A · Kimi/Claude)\n非阻塞写入"] end subgraph mp["Message Processing"] MSG["message-processor\n非阻塞写入"] end subgraph sa["Studio API (部分实施)"] STUDIO["studio-api\nClose ✅ · Postpone ❌\ntimeline / lastActivityAt / SQS gap"] end subgraph p2triggers["Contact Analysis 触发源\nV1 无 Cooldown — DB index 保护数据完整性"] CRON["❶ Cron 06:00 UTC ✅ ENABLED"] DEMAND["❷ On-demand\n后端 ✅ · 前端 DEFERRED"] FUN["❸ Per-Call Analysis SQS\n每通电话无条件触发(ai-analysis 非阻塞 send)\n✅ 已实施"] TASKCLOSE["❹ Task close ⚠️ 部分实施"] T5["T5 showed *"] T6["T6 trialed *"] end subgraph p2["Contact Analysis"] P2READ["读取\ncontacts(summary) · 增量 calls/msgs/leads\n+ tasks"] P2AI["① AI 分析 (Kimi/Claude)"] P2TEMP["② 温度计算 (纯规则)\ncomputeTemperature()\n⚠️ 待移除（从 timeline 时间戳计算）"] P2ID["②b 预生成 ID\ntaskId = randomUUID()\ncontactAnalysisRunId = sqsRecord.messageId"] end subgraph tx["db.batch() — 1 次 HTTP，原子\n⚠️ updatedAt=NOW() 必须显式传\n幂等: uq_tasks_analysis_run_category\n撞 unique → 整个 batch 回滚 → SQS 重试"] TX3["③ UPDATE contacts\ncustomerSummary · leadStatus\nlifecycleStage\nlastContactAnalysisAt=NOW()\n⚠️ temperature 待移除"] TX4["④ AI-driven task decisions 🆕\nclose: closeResult(11值) + closeNote\nupdate: priority/suggestedActions\ncreate: new tasks per AI"] TX5["⑤ taskDecisions 写入 🆕\nclose→UPDATE+timeline\ncreate→INSERT+timeline\nupdate→UPDATE+timeline"] TX6["⑥ INSERT contact_timeline 🆕\ntask.created · task.status_changed\ntask.updated · contact.lifecycle_changed"] end LEADS[("leads")] CALLS[("calls")] CONTACTS[("contacts ★\nSSoT")] MESSAGES[("messages")] TASKS[("tasks 🆕")] TIMELINE[("contact_timeline 🆕")] STAFF[("staff\nDEFERRED · 已建 · 零写入")] %% Entry → Lambda LEAD --> LT CALL --> AAP SMS --> MSG UI -.-> STUDIO %% lead-tracking writes LT -->|INSERT| LEADS LT -->|UPSERT 🆕 轻量\nON CONFLICT → updated_at\n不覆盖 AI 字段| CONTACTS LT -->|INSERT 🆕\nlead_outreach · high\n双层 dedup: per-lead + per-contact\ndueAt=SLA(5min)-quiet hours| TASKS LT -->|INSERT 🆕\nlead.created| TIMELINE %% Per-Call Analysis writes AAP -->|33 AI 字段| CALLS AAP -->|UPSERT 基础字段\nfirstName · lastName\nlastActivityAt| CONTACTS AAP -->|INSERT 🆕\ncall.created| TIMELINE AAP -->|每通电话\n无条件触发| FUN %% message-processor writes MSG -->|UPSERT| MESSAGES MSG -->|UPSERT 🆕 轻量\nlastActivityAt + updated_at| CONTACTS MSG -->|INSERT 🆕\nmessage.created| TIMELINE %% studio-api writes STUDIO -->|close ✅\npostpone ❌| TASKS STUDIO -.->|lastActivityAt\n(gap) *| CONTACTS STUDIO -.->|task.status_changed\ntask.due_at_changed (gap) *| TIMELINE STUDIO -.->|SQS → Contact Analysis\n(gap) *| TASKCLOSE %% Contact Analysis triggers CRON --> P2READ DEMAND --> P2READ FUN --> P2READ TASKCLOSE -.-> P2READ T5 -.-> P2READ T6 -.-> P2READ %% Contact Analysis reads CALLS -.->|增量| P2READ MESSAGES -.->|增量| P2READ LEADS -.->|增量| P2READ CONTACTS -.->|summary| P2READ TASKS -.->|全量| P2READ %% Contact Analysis internal flow P2READ --> P2AI --> P2TEMP --> P2ID --> TX3 %% Transaction internal flow TX3 --> TX4 --> TX5 --> TX6 %% Transaction writes to DB TX3 -->|AI 综合字段| CONTACTS TX4 -->|status=closed| TASKS TX5 -->|status=pending| TASKS TX6 -->|事件记录| TIMELINE %% Colors ``` **图例**:实线 = 写入 · 虚线 = 读取 / DEFERRED · 🆕 = Task 系统新增 · `*` = DEFERRED(V1 不实现) *** ## 一、写入模式总览 ### 1.1 所有 Lambda 统一用 Neon HTTP driver + Drizzle ORM 连接方式：`@neondatabase/serverless` HTTP driver → `drizzle-orm/neon-http`。每条 SQL 是一个 HTTP POST。 ### 1.2 多表写入：`db.batch()` vs 分开 `await` | 写法 | HTTP 请求数 | 原子性 | 用在哪 | | --------------------------------------- | :------: | :----------: | ---------------------------------- | | `await db.batch([stmt1, stmt2, stmt3])` | **1** | ✅ 全成功 or 全回滚 | 写入之间有因果依赖 | | 分开 `await db.insert(...)` | **N** | ❌ 各自独立 | 有明确"主记录"，其他是补充 | | ~~`db.transaction(async tx => {...})`~~ | — | — | ❌ **neon-http driver 不支持，会 THROW** | > `db.batch()` 底层调用 Neon 的 `sql.transaction()` — 所有 SQL 打包成 1 个 HTTP 请求，DB 内执行 BEGIN/COMMIT/ROLLBACK。 ### 1.3 五个 Flow 写入模式 所有 Flow 的 Neon 写入统一用 `db.batch()` 打包(2026-03-28 Neon-primary 决策)。DDB 写入是过渡期冗余,独立标记不参与 batch;SQS 发送在 batch 外,batch 成功后才发,非阻塞。 | Flow | Lambda | Neon 写入模式 | 原因 | | ---- | ------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | --------------------------------------------------------------------- | | 1 | lead-tracking | 3-Lambda 拆分:poller `UPSERT leads`(fail-open,失败 → neon-write-dlq) → NeonRetryProcessor retry + `LeadCreated` → EventBridge/SQS → lead-processor `db.batch([contacts, tasks, timeline])` | leads 写入失败走 DLQ 异步重试,email still SEEN;downstream 失败走 SQS retry,避免互相阻塞 | | 2 | Per-Call Analysis | `db.batch([calls, contacts, timeline])` | calls+contacts+timeline 原子;DDB calls 过渡期独立写 | | 3 | Contact Analysis | `db.batch([contacts, ...taskDecisions, timeline])` | 关旧 task + 建新 task 必须原子(否则"幽灵分析") | | 4a | studio-api Close | **raw SQL × 2**(UPDATE tasks → INSERT task\_events) | 当前未原子化,前者成功后者失败 → task 已关但无 audit;后续可改用 `db.batch()` | | 4b | studio-api Postpone | ❌ 未实施 | 暂无端点;终态用 `db.batch()` | | 5 | message-processor | per-entry `db.batch([messages, contacts, timeline])` | 每条 SQS record 一个 batch,失败 → 该 entry 进 batchItemFailures 重试 | ### 1.4 踩坑注意事项 ::: warning db.batch() 生产注意（社区踩坑总结） 1. **`db.batch()` 内不能 `Promise.all`** — 并行语句会丢 transaction scope（[Drizzle #2200](https://github.com/drizzle-team/drizzle-orm/issues/2200)） 2. **`$onUpdate(() => new Date())` 在 batch 内不生效** — Drizzle 的 `$onUpdate` 是 query builder hook，不是 DB trigger。batch 内所有 UPDATE 必须显式写 `updatedAt: sql\`NOW()\`\` 3. **`ON CONFLICT DO NOTHING` 不导致 batch 回滚** — PostgreSQL 的 conflict 处理是静默跳过，不抛错。dedup index 在 batch 内安全工作 4. **Pin driver 版本** — `@neondatabase/serverless` 和 `drizzle-orm` 之间有版本兼容性问题，社区报告升级后运行时报错（[Drizzle #5208](https://github.com/drizzle-team/drizzle-orm/issues/5208)） 5. **SQS 提供 Lambda 层重试** — `db.batch()` 解决 SQL 原子性，SQS `batchItemFailures` 解决 Lambda crash 重试。两者配合 = 幂等 + 原子 ## ::: ## 二、总览矩阵 | Lambda | 触发输入(原始数据从哪来) | AI 分析前从 Neon 读什么 | 写入哪些表 | 详情 | | ----------------- | --------------------------------------- | ----------------------------------------------------- | ---------------------------------------------- | ------------------------------------------------------------- | | lead-tracking | IMAP email（EventBridge 5min 轮询） | — | leads ✅ · contacts 🆕 · tasks 🆕 · timeline 🆕 | [详情](/system-design/write-matrix/lead-tracking-writes.md) | | Per-Call Analysis | SQS（transcript，来自 transcribe-processor） | contacts（customerSummary）+ calls(24h) + messages(24h) | calls ✅ · contacts ✅ · timeline 🆕 | [详情](/system-design/write-matrix/per-call-analysis-writes.md) | | Contact Analysis | SQS `{phone, franchiseId, siteId}` | contacts（summary）+ 增量 calls/msgs/leads + tasks | contacts ✅ · tasks 🆕 · timeline 🆕 | [详情](/system-design/write-matrix/contact-analysis-writes.md) | | message-processor | SQS（RingCentral SMS/VM webhook） | — | messages ✅ · contacts 🆕 · timeline 🆕 | [详情](/system-design/write-matrix/message-processor-writes.md) | | studio-api | API request（员工 UI 操作） | tasks（校验 status=pending）\* | tasks\* · contacts\* · timeline\* | [详情](/system-design/write-matrix/studio-api-writes.md) | ✅ = 现有 🆕 = V1 新增 \* = DEFERRED — = 不读 Neon（数据全在触发输入里） > **Per-Call Analysis 读取**：当前代码从 DDB 读（`customer-history.ts`），计划迁移到 Neon。如果 `customerSummary` 存在就读，不存在就跳过（graceful degradation）。 > > **Contact Analysis 读取**：customerSummary 是 compressed memory。`lastContactAnalysisAt` 为 NULL（首次）→ 读 30d 全量建立基线。有值 → 读 `lastContactAnalysisAt` 之后的增量。详见 [contact-analysis-writes.md](/system-design/write-matrix/contact-analysis-writes.md)。 *** ## 三、Contact Analysis 触发源 6 个触发源,统一发 SQS 消息到同一 schema(Zod-validated by `SqsMessageBodySchema`): ```typescript { phone: string, // 必填 storeId: string, // 必填 — contacts PK 组件(post-migration 0014) source?: 'cron' | 'on_demand' | 'per_call_analysis' | 'task_close' | 'showed' | 'trialed', // optional observability trace 字段 — 老 producer 不传也能解析 callId?: string, telephonySessionId?: string, analysisCompletedAt?: number, queuedAt?: number, } ``` > **`franchiseId` / `accountId` 不再传** — Consumer(contacts-analyzer)从 contact 行通过 PK lookup 自行读取。上游 producer 只需要 `phone + storeId`。 > 技术约束:SQS ESM filtering 只支持 body key(不支持 MessageAttributes)。 > > **FIFO Queue 已上线** — dailyBatchQueue 从 Standard 切换为 FIFO 并迁移到 `SqsStack`。触发源发送消息必须提供 `MessageGroupId` 和 `MessageDeduplicationId`,例如 ai-analysis-processor 用 `${phone}#${storeId}` 作 group key、`per_call_analysis-${phone}-${callId}` 作 dedup id。 | # | 触发源 | SQS Queue | `source` | 状态 | | -- | ------------------------ | --------------------------------------- | --------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | ❶ | Cron 每天 06:00 UTC | dailyBatchQueue (FIFO) | `'cron'` | ✅ ENABLED in CDK | | ❷ | On-demand（前端 Refresh AI） | dailyBatchQueue（V1）/ expressQueue\* | `'on_demand'` | ✅ 后端现有，前端 DEFERRED | | ❸ | Per-Call Analysis（每通电话） | dailyBatchQueue（V1）/ expressQueue\* | `'per_call_analysis'` | ✅ 已实施（ai-analysis-processor non-blocking SQS send） | | ❹ | Task close（员工关闭 Task） | dailyBatchQueue (FIFO) / expressQueue\* | `'task_close'` | ⚠️ 部分实施 — studio-api 已能关 task + 写 audit，timeline / contacts.lastActivityAt / SQS 触发暂未补齐（详见 [studio-api-writes.md](/system-design/write-matrix/studio-api-writes.md)） | | T5 | showed（4h 黄金窗口） | expressQueue\* | `'showed'` | 🆕 DEFERRED | | T6 | trialed（2-4h 黄金窗口） | expressQueue\* | `'trialed'` | 🆕 DEFERRED | > \* expressQueue 尚未创建（Phase D5）。V1 期间 ❷❸❹ 用 dailyBatchQueue。 > ❷ On-demand 后端 API 已存在，但前端 Refresh AI 按钮随员工 UI 一起 DEFERRED。 ### 字段来源 | 触发源 | phone / storeId 从哪来 | | ------------------- | --------------------------------------------------------------------- | | ❶ Cron | contacts 表遍历(dispatch fan-out 时发 per-contact 消息) | | ❷ On-demand | 前端 API request payload | | ❸ Per-Call Analysis | ai-analysis-processor 解析 storeId 后发送(`contactPhone + storeId` 都解析才触发) | | ❹ Task close | tasks 表的 `contact_phone` + `store_id` | | T5/T6 | Contact Analysis 自身检测到 leadStatus 变化后发送 | ### V1 不做 Cooldown 不做应用层 cooldown,重复分析由 DB index 保护(`uq_tasks_pending_contact_category` 防重复 task,`uq_tasks_analysis_run_category` 防 SQS 重试)。成本 \~200 calls/day × $0.005 ≈ $1/day,可接受。`lastContactAnalysisAt` 每次分析仍写 `NOW()`,未来加 cooldown 时可直接读取。 ### T5/T6 推荐实现(DEFERRED) Contact Analysis 自检测 leadStatus 变化 → EventBridge Scheduler 延迟触发(showed=4h, trialed=2-4h) — SQS `DelaySeconds` 最大 15min 不够用。Loop guard:`source` 已是 showed/trialed 时不再检测。 *** ## 四、contacts 表写入详情 contacts 是中心化客户档案,4 个 Lambda 各写不同字段。 ### 字段所有权 **Per-Call Analysis 只写"活动标记"**(什么时候发生了什么),**Contact Analysis 只写"AI 判断"**(客户状态是什么) — 两者字段**互不重叠**。 | 分类 | 字段 | 所有者 | | ----- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------ | | 身份 | firstName, lastName | Per-Call Analysis（初始值由 lead-tracking 设） | | 活动时间 | lastActivityAt | Per-Call Analysis + lead-tracking + message-processor + studio-api | | 业务数据 | hasCardOnFile | Per-Call Analysis（`credit_card_captured` AI 检测 → contacts 直写） | | 运营 | notes | 员工（studio-api，DEFERRED） | | 初始状态 | lifecycleStage='lead', leadStatus='new' | lead-tracking（仅 INSERT） | | AI 画像 | customerSummary, leadStatus, leadStatusReason, lifecycleStage, lifecycleState, purchaseIntent, purchaseIntentReason, goals, leadObjections, leadRejectionReasons | Contact Analysis | | AI 行动 | actionNeeded, actionNeededReason, suggestedActions | Contact Analysis | | AI 风险 | doNotContact, doNotContactUpdatedBy, hasOpenComplaint | Contact Analysis | | AI 追踪 | lastContactAnalysisAt | Contact Analysis | | 系统 | createdAt, updatedAt | DB / 所有 Lambda | ### 三个 UPSERT contacts 对比 lead-tracking / Per-Call Analysis / message-processor 都 UPSERT contacts,但写的字段不同 — 数据源决定能写什么。 **INSERT 时写入:** | 字段 | lead-tracking | Per-Call Analysis | message-processor | 差异原因 | | -------------------------- | :-----------: | :---------------: | :---------------: | ---------------------------------------------------------------------------------------------------------- | | phone, franchiseId, siteId | ✅ | ✅ | ✅ | PK，三者都需要 | | lifecycleStage='lead' | ✅ | — | — | 只有 lead-tracking 知道这是 lead | | leadStatus='new' | ✅ | — | — | 同上 | | firstName | ✅（email 解析） | ✅（call AI 提取） | — | SMS webhook 不含名字（[#521](https://github.com/retaintive/callytics-infrastructure/issues/521) 未来可从 RC API 补写） | | lastName | ✅ | ✅ | — | 同上 | | lastActivityAt | ✅ `NOW()` | ✅ `NOW()` | ✅ `NOW()` | 三者一致 — 任何互动都是活动 | | ~~firstAttemptedAt~~ | — | ~~✅~~ 待移除 | — | 通话才算"尝试联系" | | ~~firstConnectedAt~~ | — | ~~✅~~ 待移除 | — | 接通才算"首次接通" | **ON CONFLICT 时更新:** | 字段 | lead-tracking | Per-Call Analysis | message-processor | 差异原因 | | -------------- | :-------------: | :---------------: | :---------------: | ----------------------------------- | | lastActivityAt | `NOW()`（覆盖） | `NOW()`（覆盖） | `NOW()`（覆盖） | 三者一致 | | firstName | `COALESCE`（只补写） | 覆盖（call AI 更准） | — | lead email 名字不如通话准，用 COALESCE 保留已有值 | | lastName | `COALESCE` | 覆盖 | — | 同上 | | lifecycleStage | **不覆盖** | — | — | 已有 contact 由 Contact Analysis 管理 | | leadStatus | **不覆盖** | — | — | 同上 | | updatedAt | `NOW()` | `NOW()` | `NOW()` | 三者一致 | > 详见各 per-Lambda 文件的 Step ② section。 *** ## 五、contact\_timeline 表写入详情 contact\_timeline 记录所有客户互动和状态变更事件,5 个 Lambda 共写 9 种事件类型。每条都包含公共必填字段(`contactPhone` / `franchiseId` / `siteId` / `eventType` / `eventCategory` / `occurredAt` / `actorType`)。 | Lambda（文件） | Step | eventType | entityType | entityId | oldValue | newValue | occurred\_at | actorType | | -------------------------------------------------------------------------------------- | ---- | --------------------------------- | ---------- | ------------------------ | -------------------------------------------- | ---------------------------------------------------------------- | ------------------------ | ------------------ | | lead-tracking（[Step ④](/system-design/write-matrix/lead-tracking-writes.md)） | ④ | `lead.created` | lead | lead.id | — | — | **lead.receivedAt** | `lead_webhook` | | Per-Call Analysis（[Step ④](/system-design/write-matrix/per-call-analysis-writes.md)） | ④ | `call.created` | call | call.telephonySessionId | — | `{callDirection, staffName, duration}` | **call.startTime** | `call_analysis` | | message-processor（[Step ③](/system-design/write-matrix/message-processor-writes.md)） | ③ | `message.created` | message | String(message.id) | — | `{direction, messageType}` | **message.creationTime** | `system` | | Contact Analysis（[Step ⑥](/system-design/write-matrix/contact-analysis-writes.md)） | ⑥ | `task.created` | task | String(taskId) | — | `{taskType, typeCategory, priority}` | `NOW()`（tx） | `contact_analysis` | | Contact Analysis（[Step ⑥](/system-design/write-matrix/contact-analysis-writes.md)） | ⑥ | `task.status_changed` | task | String(closed\_task\_id) | `{status: pending}` | `{status: closed, closeType: auto_closed, closeResult: AI(11值)}` | `NOW()`（tx） | `contact_analysis` | | Contact Analysis（[Step ⑥](/system-design/write-matrix/contact-analysis-writes.md)） | ⑥ | `task.updated` | task | String(taskId) | `{priority: old, suggestedActions: old}` | `{priority: new, suggestedActions: new, reason}` | `NOW()`（tx） | `contact_analysis` | | Contact Analysis（[Step ⑥](/system-design/write-matrix/contact-analysis-writes.md)） | ⑥ | `contact.lifecycle_changed` | contact | — | `{lifecycleStage: old, lifecycleState: old}` | `{lifecycleStage: new, lifecycleState: new}` | `NOW()`（tx） | `contact_analysis` | | ~~Contact Analysis（[Step ⑥](/system-design/write-matrix/contact-analysis-writes.md)）~~ | ⑥ | ~~`contact.temperature_changed`~~ | contact | — | `{leadTemperature: old}` | `{leadTemperature: new, reason}` | `NOW()`（tx） | `contact_analysis` | | studio-api（[4a Step ③](/system-design/write-matrix/studio-api-writes.md)）\* | ③ | `task.status_changed` | task | String(task.taskId) | `{status: pending}` | `{status: closed, closeType: manual_closed, closeResult}` | `NOW()` | `staff` | | studio-api（[4b Step ②](/system-design/write-matrix/studio-api-writes.md)）\* | ② | `task.due_at_changed` | task | String(task.taskId) | `{dueAt: old}` | `{dueAt: new, reason}` | `NOW()` | `staff` | \* = DEFERRED > **occurred\_at 规则**:非 transaction 写入(lead / call / SMS)用**源记录时间**防排序漂移;transaction 内(Contact Analysis Step ⑥)用 `NOW()`,所有语句共享同一时间。 > > **错误处理**:5 个 Lambda 的 timeline INSERT **都在 `db.batch()` 内,batch 阻塞**(失败 → 整个 batch 回滚 → SQS 重试)。lead-tracking 的 prod Neon 写入阻塞,test Neon fire-and-forget;其他 Lambda 的 SQS send / DDB 旁路写入是非阻塞。studio-api Close 当前是 raw SQL × 2(UPDATE tasks + INSERT task\_events)未原子化;不写 contact\_timeline。 *** ## 六、字段清理计划 有了 contact\_timeline 后,contacts 表上部分字段可从 timeline 派生 → 计划移除。temperature 相关字段和 timeline 事件也将一并移除。 ### 已移除的 contacts 字段 | 字段 | 替代方案 | PR | | ------------------------- | --------------------------------------- | ------------------------------------------------------------------------------- | | ~~`leadStatusChangelog`~~ | timeline `contact.lifecycle_changed` 事件 | [callytics-common#33](https://github.com/retaintive/callytics-common/pull/33) ✅ | ### 待移除的 contacts 字段(需 engineer review,[callytics-common#32](https://github.com/retaintive/callytics-common/issues/32)) **Phase 1:timeline 有数据后可移除** | 移除字段 | 替代方案 | | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | ------------------------------------------------------------------------------------------------------------------------------------------------------- | | `acquiredAt` | `MIN(occurred_at) FROM contact_timeline WHERE event_type='lead.created'` | | `firstAttemptedAt` | `MIN(occurred_at) FROM contact_timeline WHERE event_type='call.status_changed' AND new_value->>'status'='Setup' AND new_value->>'direction'='Outbound'` | | **Phase 1 起**(infra#592 / [#681](https://github.com/retaintive/callytics-infrastructure/pull/681) merged 2026-04-29):用 `call.status_changed` Setup 事件 — 拨号瞬间记录,不等 AI 分析(原 `call.created` 是 Disconnected 后回填,失去实时性) | | | `firstConnectedAt` | `MIN(occurred_at) FROM contact_timeline WHERE event_type='call.status_changed' AND new_value->>'status'='Answered'` | | **Phase 1 起**:RC `Answered` = 路由层接通(不一定是 human conversation,见 SoT § 5)。如果只算"真说话",仍需读 `calls.callState='human_conversation'`(AI 分析填) | | | `lastComplaintAt` | timeline complaint 事件 | | `leadTemperature` / `leadTemperatureReason` | 从 timeline 到达时间实时计算(`NOW() - occurredAt`),无需存储或中间函数 | **Phase 2:Task Pipeline 完成 + 前端迁移后可移除** | 移除字段 | 替代方案 | 为什么等 | | -------------------- | ---------------------------------------------------- | -------------------------------------------------------------------------- | | `actionNeeded` | `EXISTS(SELECT 1 FROM tasks WHERE status='pending')` | 有了 tasks 表后这是 stale snapshot — 员工关了 task 但 `contacts.actionNeeded` 仍为 true | | `actionNeededReason` | tasks 表同名字段 | 跟随 `actionNeeded` | | `suggestedActions` | tasks 表同名字段(更新鲜) | 同上。需删 `idx_contacts_action_needed` | ### 待移除的 timeline 事件 ~~`contact.temperature_changed`~~ — 随 temperature 字段一起移除。 ### 为什么 timeline 比 contacts 列更好 重复进入场景(客户 3 月来 → 冷却 → 9 月再来)时,contacts 列的 `acquiredAt` 永远冻结在 3 月。Timeline 可查当前周期的 `lead.created`,准确算 Speed to Lead。 ### 相关决策变更 - **typeCategory 门控**:`temperature≠cold` → 改为 AI prompt 指令("lastActivityAt > 5d 不建 lead\_follow\_up") - **Stale override**:>30d → neglected → 改为"30 天无活动 + 无 pending task"查询 > **当前状态**:新代码(lead-tracking / message-processor)不写这些字段。Contact Analysis 现有代码仍写,等 timeline 有数据 + 验证后停止。 *** ## 七、已知问题与代码隐患 Expert Panel 代码验证发现 10 个待修复问题,按严重度排序。 | # | 问题 | 严重度 | 当前状态 | GH Issue | | -- | -------------------------------------------- | ------ | ----------------------------------------------------------------------- | ------------------------------------------------------------------------------- | | 1 | ~~`getRecentLeads` 缺 `franchiseId` 过滤~~ | ~~P0~~ | ✅ 已修复 — 查询已加 `franchiseId` 过滤 | [#518](https://github.com/retaintive/callytics-infrastructure/issues/518) | | 2 | ~~Per-Call Analysis 当前不触发 Contact Analysis~~ | ~~P0~~ | ✅ 已修复 — ai-analysis-processor 在 batch 写完后 non-blocking 发 SQS | [#519](https://github.com/retaintive/callytics-infrastructure/issues/519) | | 3 | ~~`lastContactAnalysisAt` 从未写入~~ | ~~P1~~ | ✅ 已修复 — Contact Analysis 每次写 `NOW()` | [#519](https://github.com/retaintive/callytics-infrastructure/issues/519) | | 4 | ~~`ContactData` interface 缺字段~~ | ~~P1~~ | ✅ 已修复 | [#519](https://github.com/retaintive/callytics-infrastructure/issues/519) | | 5 | Cooldown 未实现时 duplicate task | — | V1 设计不做 Cooldown — DB index `uq_tasks_pending_contact_category` 保护数据完整性 | — | | 6 | 冗余 contacts 字段清理(9 个字段) | **P2** | 待 timeline 有数据后移除 6 个 + Phase B 后移除 actionNeeded 三件套 | [callytics-common#32](https://github.com/retaintive/callytics-common/issues/32) | | 7 | ~~`writeAnalysis()` 不在 transaction~~ | ~~P1~~ | ✅ 已修复 — 唯一路径是 `writeAnalysisWithTasks()` 用 `db.batch()` 原子写入 | — | | 8 | Reserved concurrency=2 | **P2** | 多触发源时升到 5 | — | | 9 | Per-Call Analysis 从 DDB 读历史 → 迁移到 Neon | **P2** | `customer-history.ts` 改为读 Neon contacts + calls(24h) + messages(24h) | — | | 10 | Contact Analysis 30 天全量读 → 增量优化 | **P2** | customerSummary + 增量(NULL → 30d 首次,有值 → since `lastContactAnalysisAt`) | — | *** ## 八、与其他文档的关系 | 文档 | 关系 | | ---------------------------------------------------------------------------- | ------------------------- | | Task Pipeline 写入流程(`.claude/specs/2026-03-25-task-pipeline-write-flow.md`) | Pipeline 级流程设计(本目录是字段级补充) | | Expert Panel 审查报告(`.claude/specs/2026-03-25-task-pipeline-expert-review.md`) | 技术决策背景和争议记录 | | [Task 字段设计](/product-design/v1/tasks-feature/tasks-field-design.md) | tasks 表业务字段定义 | | [Contacts 数据来源与字段设计](/product-design/v1/contacts-feature/contacts-spec.md) | contacts 表字段设计来源 | | [Backend 数据管道](/architecture/2-backend.md) | 数据管道总体架构 |