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]。

  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: <LeadRow JSON>}。

前置条件(全部非 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 连接串

#Flow 1: Lead Tracking

#:::

#数据流概览

#错误处理

#Poller(src/poller.ts)

#NeonRetryProcessor(src/neon-retry-processor.ts)— 本节由 issue #201 引入

#下游 lead-processor(callytics-infrastructure)

#Duplicate LeadCreated 风险与下游 idempotency

#读取数据

#Step ①:UPSERT leads

#INSERT 时写入

#ON CONFLICT (id) DO UPDATE

#Step ②:发布 EventBridge LeadCreated 事件

#Step ③:UPSERT contacts(轻量)

#INSERT 时写入

#ON CONFLICT (phone, storeId) DO UPDATE

#不写的字段

#Step ④:INSERT tasks

#去重保护(ON CONFLICT DO NOTHING)

#Step ⑤:INSERT contact_timeline

#CDK 基础设施

#lead-tracking repo(us-east-1)

#callytics-infrastructure repo(us-west-2)

#⚠️ 已知缺失:DLQ CloudWatch alarm

#Test/Prod 双写模式

Flow 1: Lead Tracking

:::

数据流概览

错误处理

Poller(`src/poller.ts`)

NeonRetryProcessor(`src/neon-retry-processor.ts`)— 本节由 issue #201 引入

下游 lead-processor(callytics-infrastructure)

Duplicate `LeadCreated` 风险与下游 idempotency

读取数据

Step ①:UPSERT leads

INSERT 时写入

ON CONFLICT (id) DO UPDATE

Step ②:发布 EventBridge LeadCreated 事件

Step ③:UPSERT contacts(轻量)

INSERT 时写入

ON CONFLICT (phone, storeId) DO UPDATE

不写的字段

Step ④:INSERT tasks

去重保护(ON CONFLICT DO NOTHING)

Step ⑤:INSERT contact_timeline

CDK 基础设施

lead-tracking repo(us-east-1)

callytics-infrastructure repo(us-west-2)

⚠️ 已知缺失:DLQ CloudWatch alarm

Test/Prod 双写模式