> For AI agents: the complete documentation index is available at /llms.txt, the full documentation bundle is available at /llms-full.txt. # Flow 1: Lead Tracking **Repo**: `lead-tracking`(独立 repo + CDK stack) · **触发**: EventBridge 每 5 分钟 IMAP 轮询 · **架构**: 3 Lambda(poller leads-only + neon-retry-processor DLQ 重试 + lead-processor contacts/tasks/timeline)· **Neon 写入**: poller prod/test 双写,retry path 只重试 prod ::: details 字段命名约定 `accountId` = RingCentral OAuth `providerAccountId`,对应 Neon 列 `account_id`(callytics-common 0.21.0 起从 `siteId` rename)。下文遗留的 `siteId` 字眼按 `accountId` 理解。`storeId` 是 store-level isolation 的唯一隔离键(UUID),`accountId` 仅作审计。 ## ::: ## 数据流概览 不调用 AI,所有值来自 email 解析 `[EMAIL]` 或系统硬编码 `[SYSTEM]`。 ```text EventBridge 5min │ ▼ ┌─────────────────────────────────────────────────────┐ │ Poller Lambda (src/poller.ts) │ │ │ │ IMAP 轮询 → Email lead 数据 [EMAIL] │ │ StoresV2 mapping → franchiseId/accountId/storeId │ │ │ │ ┌──────────────────────────────────┐ │ │ │ [DDB] 过渡期 — 将来整块删除 │ │ │ │ PutCommand leads │ │ │ │ QueryCommand phone dedup │ │ │ └──────────────────────────────────┘ │ │ │ │ ① UPSERT leads (prod Neon) — 非阻塞 │ │ ├─ 成功 → ② 发 LeadCreated │ │ └─ 失败 → circuit breaker + 进 SQS DLQ │ │ │ │ ① UPSERT leads (test Neon) — fire-and-forget │ │ │ │ ② 发布 EventBridge LeadCreated 事件 │ │ 前置条件: phone + franchiseId + accountId + storeId │ │ 缺任一 → 跳过, 只写 leads │ └────────┬───────────────────────┬────────────────────┘ │ │ │ ① 失败 走 DLQ │ ② 成功 走 EventBridge ▼ │ source='lead-tracking' ┌──────────────────┐ │ detail-type='LeadCreated' │ neon-write-dlq │ │ │ SQS, retain 7d │ │ │ visibility 2min │ │ └────────┬─────────┘ │ │ │ ▼ │ ┌─────────────────────────────┐ │ │ NeonRetryProcessor Lambda │ │ │ (src/neon-retry-processor) │ │ │ batchSize=5 │ │ │ │ │ │ ① 重试 persistLeadPipeline │ │ │ ├─ 失败 → 抛错 → SQS redrive │ │ └─ 成功 → ② 发 LeadCreated │ │ │ │ │ │ ② 发 EventBridge LeadCreated │ │ │ ├─ FailedEntryCount>0 │ │ │ │ → 抛错 → SQS redrive │ │ │ └─ 成功 → 删除 SQS message │ │ │ │ │ │ maxReceiveCount=3 → 进终态 DLQ │ └──────────┬──────────────────┘ │ │ 3 次失败 │ ▼ │ ┌──────────────────┐ │ │ neon-write- │ │ │ final-dlq │ │ │ retain 14d │ │ │ 人工调查 │ │ └──────────────────┘ │ ▼ ┌───────────────┐ │ SQS Queue │ │ (lead-processor│ │ -queue) │ │ batchSize=1 │ │ maxRetry=3 │──► DLQ (14天) └───────┬───────┘ │ ▼ ┌─────────────────────────────────────────────────────┐ │ Lead Processor Lambda (src/lead-processor.ts) │ │ callytics-infrastructure repo, us-west-2 │ │ │ │ db.batch() — 1 次 HTTP，原子: │ │ │ │ ③ UPSERT contacts (轻量) │ │ lifecycle='lead' · leadStatus='new' │ │ ON CONFLICT (phone, storeId): 不覆盖 lifecycle │ │ │ │ ④ INSERT tasks │ │ lead_outreach · high · dueAt=SLA(5min) │ │ ON CONFLICT DO NOTHING（dedup） │ │ │ │ ⑤ INSERT timeline │ │ lead.created · occurredAt=receivedAt │ │ ON CONFLICT DO NOTHING (idempotencyKey) │ │ │ │ env Neon — 阻塞（失败 → SQS 重试） │ └─────────────────────────────────────────────────────┘ ``` > **3 Lambda 拆分理由**: > > - Poller 和 lead-processor 拆分:leads 写入失败(circuit breaker + DLQ)和 downstream 写入失败(SQS retry)解耦,避免互相阻塞。 > - NeonRetryProcessor 拆分:poller 走 fail-open 不阻塞 IMAP SEEN 标记;真正的 Neon 重试由独立 Lambda 异步消费 DLQ,降低 poller 延迟和复杂度。 > - retry path 和 happy path **必须发同一份 LeadCreated 事件**(issue #201),否则 DLQ 救回的 leads 永远没有下游 contacts/tasks/timeline。 *** ## 错误处理 ### Poller(`src/poller.ts`) | 阶段 | 失败时行为 | 重试机制 | | -------------------------------- | --------------------------------------------------- | ------------------------------------------ | | DDB PutCommand | 抛错 → email 不 mark SEEN | 下次 5min poll 重拉同一封 email | | leads UPSERT (prod Neon) | catch → 进 SQS neon-write-dlq, **email 仍 mark SEEN** | NeonRetryProcessor 异步消费 | | Circuit breaker 触发 | 连续 5 次失败 → 暂停 15 分钟,期间所有 Neon write 直接 → DLQ | 暂停期满 + 单次成功 → reset counter | | EventBridge publish (happy path) | catch → 进 SQS neon-write-dlq, **email 仍 mark SEEN** | NeonRetryProcessor 异步重发事件; leads UPSERT 幂等 | | Test Neon | catch → console.warn | 不重试,不影响 prod | ### NeonRetryProcessor(`src/neon-retry-processor.ts`)— 本节由 issue #201 引入 | 阶段 | 失败时行为 | 重试机制 | | ----------------------------------------- | ------------------------------------------------------ | -------------------------------------------------------------------- | | persistLeadPipeline 重试 | 抛 `RetryStageError(stage='neon')` → SQS redrive | maxReceiveCount=3 → 进 neon-write-final-dlq(14d 保留) | | EventBridge publish | 抛 `RetryStageError(stage='eventbridge')` → SQS redrive | 同上 | | `PutEvents` 返回 `FailedEntryCount > 0` | 显式检测后抛错 → SQS redrive | AWS SDK 在 200 OK + per-entry failure 时不抛,**必须显式检查**,否则消息被删除 = 静默丢失事件 | | 字段缺失(phone/franchiseId/accountId/storeId) | publish 跳过(`logger.info`)+ retry 不算失败 | 不重试 | ### 下游 lead-processor(callytics-infrastructure) | 阶段 | 失败时行为 | 重试机制 | | --------- | -------------- | ---------------------------------- | | ③④⑤ batch | throw → SQS 重试 | 最多 3 次,之后进 lead-processor-dlq(14d) | ### Duplicate `LeadCreated` 风险与下游 idempotency `NeonRetryProcessor` 写 Neon 成功后再发 LeadCreated。若 EventBridge publish 失败 → 整条 SQS 消息 redrive → 第二次执行 `persistLeadPipeline` + 重发 LeadCreated → **下游可能收到 2 次同一 lead 的 LeadCreated**。 下游 lead-processor 靠 3 个 idempotency 抗住: - `contacts` ON CONFLICT (phone, storeId) DO UPDATE — UPSERT 幂等 - `tasks` ON CONFLICT (sourceLeadId) DO NOTHING(`uq_tasks_source_lead`)— 同 lead 不创建第 2 个 task - `contact_timeline` ON CONFLICT (idempotencyKey) DO NOTHING — `lead.created:${row.id}` 作为 key `persistLeadPipeline` 自己也是 idempotent(`leads` ON CONFLICT DO UPDATE),双写主要影响 `syncedAt` bump,无业务副作用。 *** ## 读取数据 | 来源 | 内容 | 备注 | | ----------------- | -------------------------------------- | ------------------------ | | IMAP 轮询 | Email lead 原始数据 | EventBridge 每 5 分钟触发 | | StoresV2 DynamoDB | leadEmail → franchiseId/siteId/storeId | 冷启动 Scan,warm start 复用缓存 | *** ## Step ①:UPSERT leads Poller Lambda 执行 Drizzle upsert,PK 是 composite dedup key(`{email}#{phone}#{date}` 或 `{email}#{receivedAt}`)。 ### INSERT 时写入 | 字段 | 值 | 说明 | | ----------------------- | ------------------------------------ | ------------------------------------------ | | `id` | composite dedup key | PK，防 IMAP 重复轮询 | | `leadEmail` | email 发件人 | | | `leadType` | email 解析 | Web Lead / Online Intro 等 | | `firstName` | email body 解析（可空） | | | `lastName` | email body 解析（可空） | | | `phone` | email body 解析（可空） | 标准化为 E.164 格式（+1XXXXXXXXXX） | | `bookedDate` | email body 解析（可空） | 已预约日期 | | `bookedTime` | email body 解析（可空） | | | `emailSubject` | 原始 subject | | | `emailFrom` | 发件人地址 | | | `emailRecipient` | 收件人地址 | | | `extractedTrackingId` | 从收件人地址提取 | `lead-tracking+{ID}@...` → 映射 site | | `franchiseId` | StoresV2 → parseBrandFromFranchise() | 解析 brand 名（如 "orangeTheory"），不含 store code | | `siteId` | StoresV2.providerAccountId | RC 账户 ID | | `storeId` | StoresV2.storeId (UUID) | 唯一隔离键（Store-Level Isolation） | | `rawBody` | 完整 email body | 调试用 | | `isForwarded` | Boolean，默认 false | 是否转发 email | | `forwardedOriginalFrom` | 原始发件人（可空） | 转发 email 时 | | `forwardedOriginalTo` | 原始收件人（可空） | 转发 email 时 | | `forwardedOriginalDate` | 原始日期（可空） | 转发 email 时 | | `processedBy` | `'imap-poller'` | | | `receivedAt` | email 接收时间 | | | `syncedAt` | `defaultNow()` | | ### ON CONFLICT (id) DO UPDATE | 字段 | 值 | 说明 | | ------------- | --------------------------------------------- | -------------------------- | | `leadType` | 更新 | email 可能重新分类 | | `franchiseId` | 更新 | StoresV2 映射可能修正 | | `siteId` | 更新 | 同上 | | `storeId` | `COALESCE(leads.store_id, EXCLUDED.store_id)` | Forward-only：不用 NULL 覆盖已有值 | | `syncedAt` | `NOW()` | | *** ## Step ②:发布 EventBridge LeadCreated 事件 leads UPSERT 成功后由 poller 发布。Event payload: `{Source: 'lead-tracking', DetailType: 'LeadCreated', Detail: }`。 **前置条件**(全部非 null 才发布): | 字段 | 来源 | 为什么需要 | | ------------- | ----------------------------------------- | -------------------------------------------- | | `phone` | email 解析 | contacts PK 的一部分 | | `franchiseId` | StoresV2 映射 | tasks/审计需要 | | `accountId` | StoresV2 映射(RC OAuth `providerAccountId`) | tasks/审计需要 | | `storeId` | StoresV2 映射(UUID) | contacts PK 的一部分,store-level isolation 唯一隔离键 | 缺任一字段 → 跳过 EventBridge 发布,只写 leads 表,log `[EVENTBRIDGE] Skipping — missing required fields`。 CDK 路由:`LeadCreatedRule-${region}` Rule → SQS → Lead Processor Lambda。 *** ## Step ③:UPSERT contacts(轻量) 确保 contact 记录存在 + 设置初始状态。Lead Processor 在 `db.batch()` 内与 ④⑤ 原子执行。 ### INSERT 时写入 | 字段 | 值 | 说明 | | ---------------- | ------------------ | ------------------------------------------- | | `phone` | lead.phone | 复合 PK | | `storeId` | lead.storeId | 复合 PK（Store-Level Isolation 终态） | | `franchiseId` | lead.franchiseId | | | `accountId` | lead.accountId | RC providerAccountId（审计字段） | | `lifecycleStage` | `'lead'` | 标记为 lead | | `leadStatus` | `'new'` | 新 lead 初始状态。Contact Analysis 后续覆盖 | | `firstName` | lead.firstName（可空） | Email 解析出的名字 | | `lastName` | lead.lastName（可空） | 同上 | | `lastActivityAt` | `lead.receivedAt` | 用 lead 接收时间，**不是** `NOW()`（防 Lambda 执行时间漂移） | ### ON CONFLICT (phone, storeId) DO UPDATE ON CONFLICT 只补写身份和 activity 字段,**不写** `leadStatus` 和 `lifecycleStage`(这两个已有 contact 的字段由 Contact Analysis 管理)。 | 字段 | 值 | 说明 | | ---------------- | ----------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------- | | `lastActivityAt` | `lastActivityAtForward(lead.receivedAt)` | forward-only 语义(`COALESCE(GREATEST(existing, eventTime), eventTime)`),乱序 lead 不能把已有 lastActivity 倒推。共享 helper 跨 5 个 contacts writer 一致 | | `firstName` | `COALESCE(contacts.first_name, lead.firstName)` | 只补写,不覆盖已有值 | | `lastName` | `COALESCE(contacts.last_name, lead.lastName)` | 同上 | | `updatedAt` | `now`(lead-processor 当前时间) | | ### 不写的字段 | 字段 | 为什么不写 | | ------------------------------------------------------ | ---------------------------------------------- | | `acquiredAt` | 已移除,timeline `lead.created` 的 `occurredAt` 已记录 | | `leadTemperature` | 已移除,是 `NOW() - 到达时间` 的派生值,timeline 已记录到达时间 | | `customerSummary` / `purchaseIntent` / `goals` 等 AI 字段 | Contact Analysis 负责 | | `leadStatusReason` | Contact Analysis 负责 | | `firstAttemptedAt` / `firstConnectedAt` | Per-Call Analysis 负责(通话事件,不是 lead 到达) | *** ## Step ④:INSERT tasks Lead Processor 在 `db.batch()` 内与 ③⑤ 原子执行,创建 1 条 high-priority `lead_outreach` task。 | 字段 | 值 | 说明 | | -------------------- | ------------------------------------------------------------------------------ | ----------------------------------------------------- | | `taskId` | `crypto.randomUUID()` | 客户端预生成 | | `contactPhone` | lead.phone | 通过 `buildContactPhoneIdentityFields()` 写入 4-field 身份块 | | `franchiseId` | lead.franchiseId | 同上 helper | | `accountId` | lead.accountId | 同上 helper | | `storeId` | lead.storeId(已 guard 非空) | 同上 helper | | `taskType` | `'lead_outreach'` | 固定 | | `typeCategory` | `'lead_outreach'` | 固定 | | `sourceType` | `'lead'` | | | `sourceLeadId` | lead.id | 幂等 key | | `status` | `'pending'` | | | `priority` | `'high'` | 固定 | | `actionNeeded` | `true` | | | `actionNeededReason` | `New ${leadType} from ${firstName \|\| 'unknown'} ${lastName \|\| ''}`(trim 后) | 动态生成 | | `suggestedActions` | jsonb 模板 | 1 项 high-priority action,提示员工"5 分钟黄金窗口内回拨新 lead" | | `dueAt` | `computeDueAt('high', now, { slaMinutes: 5 })` | 排除 quiet hours | ### 去重保护(ON CONFLICT DO NOTHING) 撞任一 index 安全跳过,不报错。 | Index | 保护场景 | | ------------------------------------------------------------------------------------------------------------ | ------------------------------------ | | `uq_tasks_source_lead` (sourceLeadId) | 同一封 email 的 SQS 重试 | | `uq_tasks_pending_contact_category` (contactPhone, franchiseId, siteId, typeCategory) WHERE status='pending' | 同一 contact 已有 pending lead\_outreach | *** ## Step ⑤:INSERT contact\_timeline Lead Processor 在 `db.batch()` 内与 ③④ 原子执行。所有 timeline writer 通过共享 `buildTimelineValues()` helper(`@retaintive/common/db`)写入,统一字段约定。 | 字段 | 值 | | ---------------- | --------------------------------------- | | `contactPhone` | lead.phone | | `franchiseId` | lead.franchiseId | | `accountId` | lead.accountId | | `storeId` | lead.storeId | | `eventType` | `'lead.created'` | | `eventCategory` | `'lead'` | | `entityType` | `'lead'` | | `entityId` | lead.id | | `occurredAt` | `lead.receivedAt`(源记录时间,**非** `NOW()`) | | `actorType` | `'lead_webhook'` | | `newValue` | `{leadType, firstName, lastName}` jsonb | | `idempotencyKey` | `lead.created:${lead.id}` | ON CONFLICT DO NOTHING 通过 `idempotencyKey` 去重,SQS 重试幂等。 *** ## CDK 基础设施 资源定义在 `lib/leadtracking-stack.ts`(lead-tracking repo)和 `lib/stacks/sqs-stack.ts` / `eventbridge-stack.ts`(callytics-infrastructure repo)。 ### lead-tracking repo(us-east-1) | 资源 | 名称 | 说明 | | ---------------- | ---------------------------------- | ------------------------------------------------------------------- | | Lambda | `ImapPollerFunction-${region}` | IMAP 轮询 + leads 写入 + EventBridge 发布 | | Lambda | `NeonRetryProcessor-${region}` | 消费 neon-write-dlq, 重试 persistLeadPipeline + 发 LeadCreated (#201 修复) | | EventBridge Rule | `ImapPollerSchedule-${region}` | 每 5 分钟触发 poller | | EventBridge Rule | `LeadTrackingForwarding-${region}` | LeadCreated 跨区转发 us-east-1 → us-west-2 default bus | | SQS Queue | `neon-write-dlq-${region}` | poller 失败入口, retention=7d, visibility=2min, redrive 到 final-dlq | | SQS Queue | `neon-write-final-dlq-${region}` | retention=14d, 人工调查 | | DynamoDB | `LeadTracking-v2-${region}` | 过渡期主表,将来移除 | | IAM | `events:PutEvents` on default bus | 同时授给 poller 和 NeonRetryProcessor — 后者缺这条会 AccessDenied(#201 修复一并补) | ### callytics-infrastructure repo(us-west-2) | 资源 | 名称 | 说明 | | ---------------- | --------------------------------- | ----------------------------------------------------- | | Lambda | `LeadProcessorFunction-${region}` | SQS 消费 → contacts/tasks/timeline | | EventBridge Rule | `LeadCreatedRule-${region}` | `source=lead-tracking, detail-type=LeadCreated` → SQS | | SQS Queue | `lead-processor-queue-${region}` | visibilityTimeout=300s, retention=4d, maxRetry=3 | | SQS DLQ | `lead-processor-dlq-${region}` | retention=14d | ### ⚠️ 已知缺失:DLQ CloudWatch alarm `neon-write-dlq` 和 `neon-write-final-dlq` 当前**没有 CloudWatch Alarm** → 消息进入终态 DLQ 时静默,只能从 SQS console 看到。 对比 callytics-infrastructure `lib/stacks/monitoring-stack.ts` 已有完整 alarm pattern(`LeadProcessorDLQAlarm` / `TranscribeDLQAlarm` 等,threshold=1, 经 alarmTopic SNS → alarm-dispatcher Lambda → Lark 卡片)。lead-tracking 需要单独 issue 跟踪补 alarm。 *** ## Test/Prod 双写模式 Poller 仍同时写 prod 和 test Neon;NeonRetryProcessor 只消费 prod write DLQ 并重试 prod Neon。下游 lead-processor 在 callytics-infrastructure 里按当前 stack 的 environment 写对应 Neon,不是 prod Lambda 再 fire-and-forget 写 test。 | 组件 | Prod / 当前 env Neon | Test Neon | | ------------------------------------- | ---------------------------------------------- | ----------------------------------------------------- | | Poller (leads happy path) | 非阻塞(失败 → neon-write-dlq, email still SEEN) | fire-and-forget(失败只 log) | | NeonRetryProcessor (leads retry path) | 阻塞当前 SQS message(失败 → SQS redrive → final-dlq) | 不写 | | Lead Processor (③④⑤) | 阻塞(失败 → SQS retry) | 由 test stack 单独写 test Neon,不是 prod fire-and-forget 双写 | SSM 参数: | 参数 | 用途 | | --------------------------------------- | ------------- | | `/lead-tracking/neon-database-url` | Prod Neon 连接串 | | `/lead-tracking/neon-database-url-test` | Test Neon 连接串 |