Commit 596adcdc by luoqi

merge: fix/idle-probe-false-alarm → test(增量空转探针误报修复)

parents d4f3c58f 0ffb6e07
Pipeline #3441 failed in 0 seconds
--- ---
title: 取数说明 title: 取数说明
description: 瑞尔召回数据自助取数 —— 涉及哪些表、召回池怎么筛、关闭操作落在哪、数据范围怎么过滤。 description: 瑞尔召回数据自助取数 —— 涉及哪些表、召回池怎么筛、认领与关闭操作落在哪、数据范围怎么过滤。
icon: Database icon: Database
--- ---
...@@ -25,6 +25,7 @@ tenant_id = 'ruier-grp' ...@@ -25,6 +25,7 @@ tenant_id = 'ruier-grp'
followup_plans 召回工单(一行 = 一个患者的一次召回任务) followup_plans 召回工单(一行 = 一个患者的一次召回任务)
├─ plan_reasons 1:N 召回理由(命中了哪些临床信号) ├─ plan_reasons 1:N 召回理由(命中了哪些临床信号)
├─ plan_executions 1:N 执行记录(客服每次操作一行,关闭也在这) ├─ plan_executions 1:N 执行记录(客服每次操作一行,关闭也在这)
├─ plan_event_logs 1:N 归属/反馈账本(认领、返池、召回反馈)
└─ patients N:1 患者主数据 └─ patients N:1 患者主数据
└─ patient_profiles 1:1 合规标记(勿打扰 / 已故) └─ patient_profiles 1:1 合规标记(勿打扰 / 已故)
``` ```
...@@ -42,8 +43,8 @@ followup_plans 召回工单(一行 = 一个患者的一次召回任务) ...@@ -42,8 +43,8 @@ followup_plans 召回工单(一行 = 一个患者的一次召回任务)
| 值 | 患者数 | 说明 | | 值 | 患者数 | 说明 |
|---|---|---| |---|---|---|
| `瑞尔` | 349,385 | **PAC 业务实际服务的范围** | | `瑞尔` | 351,024 | **PAC 业务实际服务的范围** |
| `瑞泰` | 57,707 | 历史带入的少量数据,分析时建议排除 | | `瑞泰` | 64,883 | 不在业务范围内,**必须排除** |
```sql ```sql
join patients pt on pt.id = p.patient_id and pt.source_unit = '瑞尔' join patients pt on pt.id = p.patient_id and pt.source_unit = '瑞尔'
...@@ -102,6 +103,67 @@ order by p.priority_score desc; ...@@ -102,6 +103,67 @@ order by p.priority_score desc;
--- ---
## 三之二、认领怎么捞
「认领」= 客服把工单收到自己名下,认领后才能操作(提交结果 / 关闭 / 生成话术 / 打反馈)。
两个数据源:
| 源 | 是什么 | 用途 |
|---|---|---|
| `followup_plans.assignee_user_id` / `assigned_at` | 当前快照,返池即置 `NULL` | 现在谁手里有多少单 |
| `plan_event_logs` | 历史账本,append-only | 谁处理过哪些患者 |
⚠️ **统计"处理过"用 `plan_event_logs`,不要用 `assignee_user_id`** —— 后者返池后被清空。
账本 **2026-07-24 起**才有数据。
### plan_event_logs
| 字段 | 说明 |
|---|---|
| `event` | `claim` 认领 / `assign` 指派他人 / `release` 返池 / `feedback` 召回反馈 |
| `assignee_user_id` | 变更后的归属人;`release` 为 `NULL` |
| `actor_user_id` | 操作人。`claim` 时 = `assignee_user_id` |
| `held_seconds` | 本次归属持续秒数,仅 `release` 有值 |
| `reason` | `feedback` = `up` / `down` |
| `details` | JSON,事件补充(如反馈文字) |
| `plan_id` / `patient_id` | **按患者聚合用 `patient_id`** |
**某客服处理过哪些患者**
```sql
select l.actor_user_id as 客服, count(distinct l.patient_id) as 处理患者数
from plan_event_logs l
where l.host_id = '902b9698-0b38-4521-95a2-6e56f8d9efbc'
and l.event in ('claim', 'assign', 'feedback')
and l.created_at >= '2026-07-24'
group by 1 order by 处理患者数 desc;
```
**接手后放弃**(认领了但没提交任何执行)
```sql
select count(distinct l.patient_id)
from plan_event_logs l
where l.host_id = '902b9698-0b38-4521-95a2-6e56f8d9efbc'
and l.event = 'claim'
and not exists (
select 1 from plan_executions e where e.plan_id = l.plan_id
);
```
**认领时长分布**
```sql
select width_bucket(held_seconds, 0, 86400, 12) as 桶, count(*),
min(held_seconds), max(held_seconds)
from plan_event_logs
where event = 'release' and held_seconds is not null
group by 1 order by 1;
```
---
## 四、关闭操作怎么捞 ## 四、关闭操作怎么捞
回访在你们系统里做,PAC 只提供「关闭机会」做闭环。一次关闭在一个事务里写两处: 回访在你们系统里做,PAC 只提供「关闭机会」做闭环。一次关闭在一个事务里写两处:
...@@ -118,10 +180,10 @@ order by p.priority_score desc; ...@@ -118,10 +180,10 @@ order by p.priority_score desc;
| 值 | 中文 | 冷静期 | | 值 | 中文 | 冷静期 |
|---|---|---| |---|---|---|
| `no_intent` | 客户无意愿 | 60 天 | | `no_intent` | 客户无意愿 | 60 天 |
| `inaccurate` | 机会识别不准确 | 14 天 | | `inaccurate` | 机会识别不准确 | **永久** |
| `competitor` | 选择竞品机构 | 永久 | | `competitor` | 选择竞品机构 | 永久 |
| `price` | 价格因素 | 60 天 | | `price` | 价格因素 | 60 天 |
| `treated` | 已完成治疗 | 14 天 | | `treated` | 已完成治疗 | **永久** |
| `unreachable` | 无法联系客户 | 30 天 | | `unreachable` | 无法联系客户 | 30 天 |
| `other` | 其他(说明写在 `notes`) | 60 天 | | `other` | 其他(说明写在 `notes`) | 60 天 |
...@@ -197,7 +259,7 @@ group by 1, 2 order by 被判不准 desc; ...@@ -197,7 +259,7 @@ group by 1, 2 order by 被判不准 desc;
| `status` | `active` 待处理 / `assigned` 处理中 / `completed` 完成 / `abandoned` 放弃(含关闭)/ `superseded` 已被新版取代 | | `status` | `active` 待处理 / `assigned` 处理中 / `completed` 完成 / `abandoned` 放弃(含关闭)/ `superseded` 已被新版取代 |
| `contact_attempts` | 累计触达次数(患者级) | | `contact_attempts` | 累计触达次数(患者级) |
| `snoozed_until` | 冷静期截止,未到期不进池 | | `snoozed_until` | 冷静期截止,未到期不进池 |
| `assignee_user_id` / `assigned_at` | 经办人 / 认领时间 | | `assignee_user_id` / `assigned_at` | 经办人 / 认领时间。**当前快照**,返池即置 `NULL`;历史见 `plan_event_logs` |
| `recall_feedback` | 客服对"这条召回准不准"的反馈(`up`/`down`),独立于关闭操作 | | `recall_feedback` | 客服对"这条召回准不准"的反馈(`up`/`down`),独立于关闭操作 |
| `superseded_at` | 被新版取代的时间;当前版为 `NULL` | | `superseded_at` | 被新版取代的时间;当前版为 `NULL` |
| `created_at` | 工单生成时间 | | `created_at` | 工单生成时间 |
...@@ -246,6 +308,9 @@ group by 1, 2 order by 被判不准 desc; ...@@ -246,6 +308,9 @@ group by 1, 2 order by 被判不准 desc;
| 忘了 `superseded_at is null` | 查当前态一律加,否则同一患者多版工单重复计数 | | 忘了 `superseded_at is null` | 查当前态一律加,否则同一患者多版工单重复计数 |
| 用 `target_clinic_id in (...)` | 会漏掉集团池,要 `or target_clinic_id is null` | | 用 `target_clinic_id in (...)` | 会漏掉集团池,要 `or target_clinic_id is null` |
| 拿 `followup_plans` 数关闭量 | 关闭数查 `plan_executions`,一个工单可能有多次操作 | | 拿 `followup_plans` 数关闭量 | 关闭数查 `plan_executions`,一个工单可能有多次操作 |
| 用 `assignee_user_id` 数"处理过多少患者" | 那是快照,返池即清空;用 `plan_event_logs` |
| 按 `plan_id` 聚合认领 / 执行 | 升版本会换新 `plan_id`,同一次认领被算多次;**按 `patient_id` 聚合** |
| 用 `external_id` 对号 | 用 `medical_record_number` | | 用 `external_id` 对号 | 用 `medical_record_number` |
| 按 `created_at` 直接当北京时间 | 差 8 小时,要 `at time zone 'Asia/Shanghai'` | | 按 `created_at` 直接当北京时间 | 差 8 小时,要 `at time zone 'Asia/Shanghai'` |
| 把 36500 天后的 `snoozed_until` 当脏数据 | 那是"永久"哨兵值 | | 把 36500 天后的 `snoozed_until` 当脏数据 | 那是"永久"哨兵值 |
| 忘了排除 `source_unit='瑞泰'` | 不在业务范围 |
...@@ -22,6 +22,8 @@ icon: Terminal ...@@ -22,6 +22,8 @@ icon: Terminal
**分支规约**:`main` = 生产(只接受合并,不直接改)· `test` = 测试(可聚合测试专用代码,**永不回流 main**)· 开发分支 → main。 **分支规约**:`main` = 生产(只接受合并,不直接改)· `test` = 测试(可聚合测试专用代码,**永不回流 main**)· 开发分支 → main。
部署脚本**不写死分支**,部署的是"该机当前 checkout 的分支",所以生产机停在 main、测试机停在 test 即可。 部署脚本**不写死分支**,部署的是"该机当前 checkout 的分支",所以生产机停在 main、测试机停在 test 即可。
需要临时部署别的分支(或首次把某台机切到某分支)用 `DEPLOY_BRANCH` 覆盖:
`DEPLOY_BRANCH=test bash deploy/deploy-prod.sh`。
--- ---
...@@ -192,9 +194,17 @@ $DC ps ...@@ -192,9 +194,17 @@ $DC ps
# 回滚:切回上一个 commit 重新部署 # 回滚:切回上一个 commit 重新部署
cd /data/pac && git log --oneline -5 # 找到要回滚的 commit cd /data/pac && git log --oneline -5 # 找到要回滚的 commit
git checkout <commit> && COMPOSE_MANAGED=1 bash deploy/deploy-prod.sh git checkout <commit>
COMPOSE_MANAGED=1 bash deploy/deploy-prod.sh --no-pull # ⚠️ 必须带 --no-pull
``` ```
<Callout type="warn">
回滚**一定要带 `--no-pull`**:不带的话脚本会先 `git fetch + pull` 当前分支,
把你刚 checkout 的旧 commit 又拉回分支最新 —— 等于回滚没生效(旧版脚本会**静默**发生;
现版本在 detached HEAD 下会取到分支名 `HEAD` 导致 fetch 失败退出,至少不会假成功)。
回滚完记得 `git checkout main` 切回来,否则下次部署仍在 detached HEAD。
</Callout>
**数据层回滚**:改 yaml 口径的可以改回去再 reparse 一遍(`rawPayload` 一直保留,可逆); **数据层回滚**:改 yaml 口径的可以改回去再 reparse 一遍(`rawPayload` 一直保留,可逆);
删除类操作不可逆,执行前务必先 `--dry-run` 或 `select count(*)` 确认范围。 删除类操作不可逆,执行前务必先 `--dry-run` 或 `select count(*)` 确认范围。
......
...@@ -3,6 +3,7 @@ import { Prisma } from '@prisma/client'; ...@@ -3,6 +3,7 @@ import { Prisma } from '@prisma/client';
import { PrismaService } from '../../../prisma/prisma.service'; import { PrismaService } from '../../../prisma/prisma.service';
import { TreatmentInitiationRecallScenario } from './scenarios/treatment-initiation-recall.scenario'; import { TreatmentInitiationRecallScenario } from './scenarios/treatment-initiation-recall.scenario';
import type { PlanScenarioPlugin, ScenarioHit, ScenarioScope } from './scenario.interface'; import type { PlanScenarioPlugin, ScenarioHit, ScenarioScope } from './scenario.interface';
import { reasonNeedsRefresh } from './reason-refresh';
/** /**
* PlanEngineService — 跑所有 scenario plugin,产 FollowupPlan + PlanReason * PlanEngineService — 跑所有 scenario plugin,产 FollowupPlan + PlanReason
...@@ -380,6 +381,9 @@ export class PlanEngineService { ...@@ -380,6 +381,9 @@ export class PlanEngineService {
// 不该升版本,就地改即可 —— 否则存量 plan 永远停在旧诊所,重算修不好)。 // 不该升版本,就地改即可 —— 否则存量 plan 永远停在旧诊所,重算修不好)。
const patch: Prisma.FollowupPlanUpdateInput = {}; const patch: Prisma.FollowupPlanUpdateInput = {};
if (latest.priorityScore !== newPriorityScore) patch.priorityScore = newPriorityScore; if (latest.priorityScore !== newPriorityScore) patch.priorityScore = newPriorityScore;
// goal 是静态文案(不含天数)→ 直接比。算法改了措辞(如高龄缺牙 goal 改「活动义齿为主」)
// 要能落到存量 plan,否则详情页「本次目标」永远是旧话。
if ((latest.goal ?? null) !== (head.goal ?? null)) patch.goal = head.goal ?? null;
if (latest.targetClinicId !== targetClinicId) { if (latest.targetClinicId !== targetClinicId) {
patch.targetClinicId = targetClinicId; patch.targetClinicId = targetClinicId;
// ⭐ 归属漂移 + 已认领 → 顺带返池:诊所隔离是硬边界,单子漂出原认领人 scope 后 // ⭐ 归属漂移 + 已认领 → 顺带返池:诊所隔离是硬边界,单子漂出原认领人 scope 后
...@@ -396,8 +400,58 @@ export class PlanEngineService { ...@@ -396,8 +400,58 @@ export class PlanEngineService {
(latest.status === 'assigned' ? '(原 assigned → 返池)' : ''), (latest.status === 'assigned' ? '(原 assigned → 返池)' : ''),
); );
} }
// ⭐ reason 就地刷新(2026-07):(scenario, subKey) 没变、但 **signals 内容变了** 的算法升级
// ——典型是缺牙按年龄排治疗(signals 新增 focusCategory / patientAge)——
// 原先对存量 plan 永远不生效(生产 3,170 条高龄缺牙仍显示「目标·种植」)。
// 这里就地改:**不升版本、不动认领/状态/抑制窗/触达计数**。
// 判据见 reason-refresh.ts:只比 signals(去 daysSince)+ evidence,
// 否则天数天天变会让每日重算把全部 plan_reasons 重写一遍。
const hitByKey = new Map(usableHits.map((h) => [key(h.scenarioKey, h.subKey), h]));
const staleRows = latest.reasons.filter((r) => {
// 已关闭的 reason 是历史,不再刷新(也不会被"复活",closedReason 本就不在更新字段里)
if (r.closedReason != null) return false;
const h = hitByKey.get(key(r.scenario, r.subKey));
if (h == null) return false;
return reasonNeedsRefresh(
{ signals: r.signals, evidence: r.evidence },
{ signals: h.signals ?? null, evidence: { factIds: h.evidence.factIds } },
);
});
if (Object.keys(patch).length > 0 || staleRows.length > 0) {
await this.prisma.$transaction(async (tx) => {
if (Object.keys(patch).length > 0) { if (Object.keys(patch).length > 0) {
await this.prisma.followupPlan.update({ where: { id: latest.id }, data: patch }); await tx.followupPlan.update({ where: { id: latest.id }, data: patch });
}
for (const r of staleRows) {
const h = hitByKey.get(key(r.scenario, r.subKey))!;
await tx.planReason.update({
where: { id: r.id },
// 语义已变 → 整行重写(天数等易变字段顺带刷新,反正这次要写)。
// ⚠️ 刻意不动:closedReason / closedAt(关闭状态)、source / sourceActorId /
// campaignId(来源溯源)、lifecycle(由 scenario+subKey 决定,未变)。
data: {
reason: h.reason,
priorityScore: h.priorityScore,
signals: h.signals
? (h.signals as unknown as Prisma.InputJsonValue)
: Prisma.JsonNull,
evidence: { factIds: h.evidence.factIds } as Prisma.InputJsonValue,
breakdown: h.priorityBreakdown
? ({
priority: { ...h.priorityBreakdown },
subKey: h.subKey ?? null,
} as Prisma.InputJsonValue)
: Prisma.JsonNull,
},
});
}
});
if (staleRows.length > 0) {
this.logger.log(
`plan ${latest.id} reason 就地刷新 ${staleRows.length} (signals 语义变化,不升版本)`,
);
}
} }
return 'unchanged'; return 'unchanged';
} }
......
/**
* unchanged 分支的「reason 就地刷新」判定。
*
* ── 解决什么问题 ──
* 引擎的变化判定只比 `(scenario, subKey)` 集合;集合没变就走 unchanged 分支,**plan_reasons 一个字都不改**。
* 于是"只改了 signals 内容"的算法升级(如 2026-07 缺牙按年龄排治疗:signals 新增
* focusCategory / patientAge)对**存量 plan 永远不生效** —— 生产实测 3,170 条高龄缺牙
* 仍显示「目标·种植」。要修只能删库重算(丢认领、丢版本号)或写一次性 SQL 回填(逻辑重复)。
* 本模块让 unchanged 分支能就地刷新:**不升版本、不动认领/状态/抑制窗**。
*
* ── 为什么必须"只在语义变化时才写" ──
* reason 文案与 signals 里都内嵌**天数**(`${days} 天前` / `signals.daysSince`),每天都在变。
* 无条件刷新 = 每日重算把全部 plan_reasons 重写一遍(生产 23 万+ plan),纯废写 + WAL 膨胀。
*
* ── 判据:比 signals(去掉 daysSince)+ evidence,**不解析文案** ──
* reason 文案完全由 signals 派生 —— 诊断名 ← triggers[].code、牙位 ← toothPosition、
* 治疗类目 ← expectedCategories + patientAge(年龄适配后)。
* 所以「signals 去掉 daysSince 后相同」⇒「文案只可能差在天数上」。
* 据此判定既充分,又不依赖文案格式(将来 scenario 改文案模板不会引发误刷)。
*/
/**
* 规范化序列化:**递归按键名排序**后再 stringify。
*
* ⚠️ 必须这么做 —— Postgres `jsonb` **不保留键序**(按键长度+字节序重排存储),
* 而代码里构造 signals 的顺序是源码顺序。直接 JSON.stringify 比较,库里读出来的
* 和新算出来的键序必然不同 → 每次都判"变了" → 每日重算把全部 plan_reasons 重写一遍,
* 恰好是本模块要避免的事(本地实测:不加这层,连跑三次每次都刷新)。
*/
function canonical(v: unknown): string {
const norm = (x: unknown): unknown => {
if (Array.isArray(x)) return x.map(norm);
if (x !== null && typeof x === 'object') {
const o = x as Record<string, unknown>;
return Object.keys(o)
.sort()
.reduce<Record<string, unknown>>((acc, k) => {
if (o[k] !== undefined) acc[k] = norm(o[k]);
return acc;
}, {});
}
return x;
};
return JSON.stringify(norm(v));
}
/** 稳定投影:剔除随时间必变、不代表语义变化的字段 */
function stableSignals(signals: unknown): unknown {
if (signals == null || typeof signals !== 'object') return signals ?? null;
// daysSince 是生成那刻的快照,展示层用 signalOccurredAt 实时重算(见 applyLiveDays),
// 它变化不代表召回语义变了 → 不作为刷新依据。
const { daysSince: _drop, ...rest } = signals as Record<string, unknown>;
return rest;
}
/** evidence.factIds 顺序在不同批次可能不同,排序后再比,避免顺序抖动触发误刷 */
function stableEvidence(evidence: unknown): unknown {
if (evidence == null || typeof evidence !== 'object') return null;
const ids = (evidence as { factIds?: unknown }).factIds;
return Array.isArray(ids) ? [...ids].map(String).sort() : null;
}
export interface ReasonContent {
signals: unknown;
evidence: unknown;
}
/**
* 该 reason 行是否需要就地刷新。
* true = 语义内容变了(signals 或证据集变了)→ 整行重写(含天数等易变字段,反正要写)。
* false = 只是天数漂移 → 不写。
*/
export function reasonNeedsRefresh(oldRow: ReasonContent, nextHit: ReasonContent): boolean {
return (
canonical(stableSignals(oldRow.signals)) !== canonical(stableSignals(nextHit.signals)) ||
canonical(stableEvidence(oldRow.evidence)) !== canonical(stableEvidence(nextHit.evidence))
);
}
...@@ -182,8 +182,17 @@ export class ClickHouseSourceService { ...@@ -182,8 +182,17 @@ export class ClickHouseSourceService {
if (!cfg.cursorValue) continue; if (!cfg.cursorValue) continue;
const sql = source.queries[tableName]; const sql = source.queries[tableName];
const m = sql?.match(/FROM\s+([\w.]+)/i); const m = sql?.match(/FROM\s+([\w.]+)/i);
if (!m) continue; if (!m || !sql) continue;
const q = `SELECT count(*) AS c FROM ${m[1]} WHERE ${cfg.cursorColumn} > '${cfg.cursorValue.replace(/'/g, "''")}'`; // ⭐ 探针必须与**真实增量拉取同一套 WHERE**:游标 + 业务过滤(去 cohort)。
// 只拼游标会数进"游标之后、但被业务条件过滤掉"的行(退费单 settlement_status≠1、
// 反向结算 is_refund=1、无就诊记录患者 last_visit_time NULL)→ 真实拉取正确地不拉,
// 探针却数到 → 误报"增量空转"CRITICAL(2026-07-25 实测:settlement 正/退表各 5788
// 完全相等,正是这批全为退费单的铁证)。
const clauses = [
`${cfg.cursorColumn} > '${cfg.cursorValue.replace(/'/g, "''")}'`,
...this.extractBusinessFilters(sql),
];
const q = `SELECT count(*) AS c FROM ${m[1]} WHERE ${clauses.join(' AND ')}`;
const rs = await client.query({ query: q, format: 'JSONEachRow' }); const rs = await client.query({ query: q, format: 'JSONEachRow' });
const rows = (await rs.json()) as Array<{ c: string | number }>; const rows = (await rs.json()) as Array<{ c: string | number }>;
out[tableName] = Number(rows[0]?.c ?? 0); out[tableName] = Number(rows[0]?.c ?? 0);
......
...@@ -1074,10 +1074,22 @@ export class ColdImportService { ...@@ -1074,10 +1074,22 @@ export class ColdImportService {
// ── 增量空转探针(2026-06-10 游标格式 bug 空转三天才被发现的教训)── // ── 增量空转探针(2026-06-10 游标格式 bug 空转三天才被发现的教训)──
// fetched=0 且确实带游标跑的增量 → 反查 DW「比游标新的行数」: // fetched=0 且确实带游标跑的增量 → 反查 DW「比游标新的行数」:
// DW 也是 0 → 真没新数据,静默;DW > 0 → PAC 拉不到但 DW 有 → 当天告警(企微)。 // DW 也是 0 → 真没新数据,静默;DW > 0 → PAC 拉不到但 DW 有 → 当天告警(企微)。
//
// ⭐ 只在 **status=success** 的空轮上探(2026-07-25 修):
// 失败轮(如 ClickHouse 连接中途断 "unexpected end of file")本就 tx+dup=0,且会经
// 日报失败计数 + error_message 暴露。在失败轮上跑探针,会把**瞬时网络失败误诊成
// "游标空转/格式失效"** —— 实测就发生过:一轮 failed,下一轮 tx=10万+ 全额补回(游标
// 未推进,无数据丢失),却先收到一条误导性的 CRITICAL 空转告警。
const usedCursor = const usedCursor =
incrementalConfig && incrementalConfig &&
Object.values(incrementalConfig.perQuery).some((c) => c.cursorValue !== null); Object.values(incrementalConfig.perQuery).some((c) => c.cursorValue !== null);
if (!options.dryRun && usedCursor && totals.transactionsWritten + totals.duplicates === 0 && manifest.sql_source) { if (
!options.dryRun &&
status === SyncStatus.SUCCESS &&
usedCursor &&
totals.transactionsWritten + totals.duplicates === 0 &&
manifest.sql_source
) {
const counts = await this.chSource.probeNewRowCounts( const counts = await this.chSource.probeNewRowCounts(
manifest.sql_source, manifest.sql_source,
incrementalConfig!.perQuery, incrementalConfig!.perQuery,
......
import { ClickHouseSourceService } from '../src/modules/sync/cold-import/clickhouse-source.service';
/**
* 增量空转探针回归。
*
* 探针职责:增量成功但拉 0 行时,反查 DW「比游标新的行数」,分辨
* 「DW 真没新数据(静默)」vs「游标条件失效在空转(CRITICAL 告警)」。
*
* 2026-07-25 生产误报排查暴露两个 bug:
* ① 探针反查只拼游标、**不带业务过滤**,把"游标之后但会被业务条件过滤掉"的行也数进去
* —— 退费单(settlement_status≠1 / is_refund=1)首当其冲。实测 settlement 正/退表各 5788
* 完全相等,正是这批全为退费单的铁证 → 真实拉取正确地不拉,探针却告警。
* ② 探针在 **失败轮** 上也跑(条件只看 tx+dup=0,没看 status),把瞬时网络失败
* ("unexpected end of file")误诊成"游标空转"。② 在 cold-import.service 的调用点修
* (加 status===success 守卫),① 在此处的 extractBusinessFilters 修。
*
* 本文件锁 ①:探针拼的 WHERE 必须与真实增量拉取同一套业务过滤。
*/
// extractBusinessFilters 是纯字符串逻辑、无构造依赖,直接实例化用 bracket 访问。
const svc = new ClickHouseSourceService();
const extract = (sql: string): string[] =>
(svc as unknown as { extractBusinessFilters(s: string): string[] }).extractBusinessFilters(sql);
describe('extractBusinessFilters — 探针复用真实拉取的业务过滤', () => {
test('⭐ 结算正表:保留 is_refund / settlement_status,去掉 cohort', () => {
const sql = `
SELECT id, patient_id, settlement_money
FROM dw_group.fact_settlement_out
WHERE (is_refund IS NULL OR is_refund = 0)
AND settlement_status = 1
AND (patient_id, brand) IN (
SELECT patient_id, brand FROM dw_group.fact_client_out WHERE last_visit_time IS NOT NULL
)`;
const f = extract(sql);
expect(f).toContain('settlement_status = 1');
expect(f.some((c) => /is_refund/.test(c))).toBe(true);
// cohort 子查询必须被剔除(否则探针 SQL 里嵌一坨子查询,且语义错)
expect(f.some((c) => /\(patient_id\s*,\s*brand\)\s+IN/i.test(c))).toBe(false);
});
test('⭐ 结算退费表:保留退费判定,去掉 cohort', () => {
const sql = `
SELECT id, patient_id
FROM dw_group.fact_settlement_out
WHERE (is_refund = 1 OR settlement_status = 4)
AND (patient_id, brand) IN (SELECT patient_id, brand FROM dw_group.fact_client_out WHERE last_visit_time IS NOT NULL)`;
const f = extract(sql);
expect(f.some((c) => /is_refund = 1 OR settlement_status = 4/.test(c))).toBe(true);
expect(f.some((c) => /\(patient_id\s*,\s*brand\)\s+IN/i.test(c))).toBe(false);
});
test('结算方式表:保留 settlement_status=1', () => {
const sql = `SELECT * FROM dw_group.fact_settlement_mode_out WHERE settlement_status = '1'
AND (patient_id, brand) IN (SELECT patient_id, brand FROM dw_group.fact_client_out WHERE last_visit_time IS NOT NULL)`;
expect(extract(sql)).toContain("settlement_status = '1'");
});
test('主档表:last_visit_time IS NOT NULL 被剔除(增量按 cursor 拉,该条冗余)', () => {
const sql = `SELECT * FROM dw_group.fact_client_out WHERE last_visit_time IS NOT NULL`;
expect(extract(sql)).toEqual([]);
});
test('预约/病历:整段 WHERE 只有 cohort → 探针无附加过滤(与真实拉取一致)', () => {
const sql = `SELECT * FROM dw_group.fact_appointment_out
WHERE (patient_id, brand) IN (SELECT patient_id, brand FROM dw_group.fact_client_out WHERE last_visit_time IS NOT NULL)`;
expect(extract(sql)).toEqual([]);
});
test('括号内的 OR/AND 不被 AND 拆开(退费表 (a OR b) 整体保留)', () => {
const sql = `SELECT * FROM t WHERE (is_refund = 1 OR settlement_status = 4) AND settlement_status = 1`;
const f = extract(sql);
expect(f).toContain('(is_refund = 1 OR settlement_status = 4)');
expect(f).toContain('settlement_status = 1');
expect(f).toHaveLength(2);
});
});
...@@ -12,7 +12,15 @@ const HOST = 'host-1'; ...@@ -12,7 +12,15 @@ const HOST = 'host-1';
const TENANT = 'tenant-1'; const TENANT = 'tenant-1';
const SCEN = 'treatment_initiation_recall'; const SCEN = 'treatment_initiation_recall';
type Reason = { scenario: string; subKey: string | null }; // id / evidence / signals:unchanged 分支的 reason 就地刷新要用(见 reason-refresh.ts)
type Reason = {
scenario: string;
subKey: string | null;
id?: string;
evidence?: unknown;
signals?: unknown;
closedReason?: string | null;
};
interface Plan { interface Plan {
id: string; id: string;
hostId: string; hostId: string;
...@@ -156,17 +164,24 @@ function makeStore(seed: { plans?: Partial<Plan>[]; personas?: { patientId: stri ...@@ -156,17 +164,24 @@ function makeStore(seed: { plans?: Partial<Plan>[]; personas?: { patientId: stri
}), }),
}; };
// unchanged 分支的 reason 就地刷新会用到(见 reason-refresh.ts)
const planReason = { update: jest.fn(async (_args: unknown) => ({})) };
const prisma = { const prisma = {
followupPlan, followupPlan,
planReason,
persona, persona,
planGenerationLog, planGenerationLog,
// 最后到诊诊所查询(prefetchForBatch):本套件不测归属 → 返空 = 回退兜底 head.targetClinicId // 最后到诊诊所查询(prefetchForBatch):本套件不测归属 → 返空 = 回退兜底 head.targetClinicId
$queryRaw: jest.fn(async (): Promise<Array<{ patient_id: string; clinic_id: string }>> => []), $queryRaw: jest.fn(async (): Promise<Array<{ patient_id: string; clinic_id: string }>> => []),
$transaction: jest.fn(async (cb: (tx: unknown) => Promise<unknown>) => $transaction: jest.fn(async (cb: (tx: unknown) => Promise<unknown>) =>
cb({ followupPlan: { update: followupPlan.update, create: followupPlan.create } }), cb({
followupPlan: { update: followupPlan.update, create: followupPlan.create },
planReason: { update: planReason.update },
}),
), ),
}; };
return { prisma, plans, logs }; return { prisma, plans, logs, planReason };
} }
function makeScenario(hits: ScenarioHit[]) { function makeScenario(hits: ScenarioHit[]) {
...@@ -226,6 +241,88 @@ describe('runAllForHost 批量路径 — 5 种结局等价', () => { ...@@ -226,6 +241,88 @@ describe('runAllForHost 批量路径 — 5 种结局等价', () => {
expect(plans.find((p) => p.id === 'p-old')!.priorityScore).toBe(55); // 刷分 expect(plans.find((p) => p.id === 'p-old')!.priorityScore).toBe(55); // 刷分
}); });
// ⭐ 2026-07:(scenario,subKey) 没变、但 signals 语义变了(如缺牙按年龄排治疗新增
// focusCategory / patientAge)—— 原先走 unchanged 分支一个字都不改,存量 plan 永远修不好。
test('unchanged + signals 语义变化 → reason 就地刷新,**不升版本、不动认领**', async () => {
const { prisma, plans, planReason } = makeStore({
plans: [
{
id: 'p-stale',
patientId: 'pat-age',
version: 3,
status: 'assigned',
assigneeUserId: 'u-cs',
contactAttempts: 2,
priorityScore: 50,
reasons: [
{
id: 'r-stale',
scenario: SCEN,
subKey: 'missing_tooth@16',
evidence: { factIds: ['f1'] },
// 旧 signals:没有年龄适配字段
signals: { subKey: 'missing_tooth', expectedCategories: ['implant', 'prosthodontic'] },
},
],
},
],
});
const h = hit('pat-age', 'missing_tooth@16', 50);
// 新 signals 带上年龄适配结果
(h as { signals?: unknown }).signals = {
subKey: 'missing_tooth',
expectedCategories: ['implant', 'prosthodontic'],
focusCategory: 'prosthodontic',
patientAge: 90,
};
const res = await engine(prisma, makeScenario([h])).runAllForHost({
hostId: HOST, tenantId: TENANT, now: NOW,
});
expect(res.plansUnchanged).toBe(1);
expect(res.plansSuperseded).toBe(0);
expect(plans.filter((p) => p.patientId === 'pat-age')).toHaveLength(1); // 没升版本
const p = plans.find((x) => x.id === 'p-stale')!;
expect(p.version).toBe(3); // 版本号未动
expect(p.status).toBe('assigned'); // 状态未动
expect(p.assigneeUserId).toBe('u-cs'); // ⭐ 认领未丢
expect(p.contactAttempts).toBe(2); // 触达计数未清
// reason 行被就地重写
expect(planReason.update).toHaveBeenCalledTimes(1);
const arg = planReason.update.mock.calls[0]![0] as {
where: { id: string }; data: { signals: Record<string, unknown> };
};
expect(arg.where.id).toBe('r-stale');
expect(arg.data.signals.focusCategory).toBe('prosthodontic');
expect(arg.data.signals.patientAge).toBe(90);
});
test('⭐ unchanged + signals 完全一致 → 不写 reason(防每日重算全量重写)', async () => {
const signals = {
subKey: 'missing_tooth',
expectedCategories: ['implant', 'prosthodontic'],
focusCategory: 'prosthodontic',
patientAge: 90,
daysSince: 100,
};
const { prisma, planReason } = makeStore({
plans: [
{
id: 'p-fresh', patientId: 'pat-same', version: 1, status: 'active', priorityScore: 50,
reasons: [
{ id: 'r1', scenario: SCEN, subKey: 'missing_tooth@16',
evidence: { factIds: ['f1'] }, signals },
],
},
],
});
const h = hit('pat-same', 'missing_tooth@16', 50);
// 只有天数往前走了一天 —— 语义没变
(h as { signals?: unknown }).signals = { ...signals, daysSince: 101 };
await engine(prisma, makeScenario([h])).runAllForHost({ hostId: HOST, tenantId: TENANT, now: NOW });
expect(planReason.update).not.toHaveBeenCalled();
});
test('superseded:既有 active 不同 subKey → plansSuperseded=1,旧 superseded + 新 v2 active', async () => { test('superseded:既有 active 不同 subKey → plansSuperseded=1,旧 superseded + 新 v2 active', async () => {
const { prisma, plans } = makeStore({ const { prisma, plans } = makeStore({
plans: [ plans: [
......
import { reasonNeedsRefresh } from '../src/modules/plan/engine/reason-refresh';
/**
* unchanged 分支「reason 就地刷新」判定回归。
*
* 背景:引擎变化判定只比 (scenario, subKey) 集合,集合没变就走 unchanged 分支、plan_reasons 一字不改。
* 于是"只改 signals 内容"的算法升级对存量 plan 永远不生效 —— 生产 3,170 条高龄缺牙即使
* 上线了按年龄排治疗,仍显示「目标·种植」。
*
* 修复引入的**新风险**(本文件主要防这个):reason 文案与 signals 都内嵌天数,天天变。
* 若无条件刷新,每日重算会把全部 plan_reasons(生产 23 万+ plan)重写一遍 —— 纯废写。
* 所以判据必须**忽略纯时间漂移**,只认语义变化。
*/
const BASE = {
subKey: 'missing_tooth',
triggers: [{ type: 'diagnosis', code: 'K08' }],
toothPosition: '16;17',
daysSince: 118,
signalOccurredAt: '2026-03-28T00:00:00.000Z',
expectedCategories: ['implant', 'prosthodontic'],
};
const EV = { factIds: ['f1', 'f2'] };
const row = (signals: unknown, evidence: unknown = EV) => ({ signals, evidence });
describe('reasonNeedsRefresh — 只在语义变化时才写', () => {
test('⭐ 完全相同 → 不刷新', () => {
expect(reasonNeedsRefresh(row({ ...BASE }), row({ ...BASE }))).toBe(false);
});
test('⭐ 只有 daysSince 变(天天都会发生)→ **不刷新**,否则每日重算全量重写', () => {
expect(reasonNeedsRefresh(row({ ...BASE, daysSince: 118 }), row({ ...BASE, daysSince: 119 }))).toBe(
false,
);
// 跨很多天也一样
expect(reasonNeedsRefresh(row({ ...BASE, daysSince: 1 }), row({ ...BASE, daysSince: 999 }))).toBe(
false,
);
});
test('⭐ 新增 focusCategory / patientAge(本次年龄适配)→ 刷新', () => {
const next = { ...BASE, focusCategory: 'prosthodontic', patientAge: 90 };
expect(reasonNeedsRefresh(row({ ...BASE }), row(next))).toBe(true);
});
test('⭐ focusCategory 改变(如患者过生日跨过 70 岁线)→ 刷新', () => {
const before = { ...BASE, focusCategory: 'implant', patientAge: 70 };
const after = { ...BASE, focusCategory: 'prosthodontic', patientAge: 71 };
expect(reasonNeedsRefresh(row(before), row(after))).toBe(true);
});
test('牙位变化 → 刷新(新长出缺牙,文案要跟着变)', () => {
expect(reasonNeedsRefresh(row({ ...BASE }), row({ ...BASE, toothPosition: '16;17;27' }))).toBe(true);
});
test('expectedCategories 变化 → 刷新(排除闸口径变了)', () => {
expect(
reasonNeedsRefresh(row({ ...BASE }), row({ ...BASE, expectedCategories: ['prosthodontic'] })),
).toBe(true);
});
test('触发信号锚点变化 → 刷新(换了更早/更晚的诊断)', () => {
expect(
reasonNeedsRefresh(row({ ...BASE }), row({ ...BASE, signalOccurredAt: '2026-01-01T00:00:00.000Z' })),
).toBe(true);
});
test('⭐ 旧 plan 无 signals(历史数据)→ 刷新,让它补上', () => {
expect(reasonNeedsRefresh(row(null), row({ ...BASE }))).toBe(true);
});
test('新旧都无 signals → 不刷新(没东西可补)', () => {
expect(reasonNeedsRefresh(row(null), row(null))).toBe(false);
});
});
describe('reasonNeedsRefresh — evidence 维度', () => {
test('证据集变化(命中新 fact)→ 刷新', () => {
expect(reasonNeedsRefresh(row(BASE, { factIds: ['f1'] }), row(BASE, { factIds: ['f1', 'f3'] }))).toBe(
true,
);
});
test('⭐ factIds 只是顺序不同 → **不刷新**(批次间顺序会抖,不能当变化)', () => {
expect(
reasonNeedsRefresh(row(BASE, { factIds: ['f2', 'f1'] }), row(BASE, { factIds: ['f1', 'f2'] })),
).toBe(false);
});
test('evidence 缺失 / 形状异常不炸,按无证据处理', () => {
expect(reasonNeedsRefresh(row(BASE, null), row(BASE, null))).toBe(false);
expect(reasonNeedsRefresh(row(BASE, {}), row(BASE, null))).toBe(false);
});
});
describe('⭐ 红线:jsonb 键序不能被当成变化', () => {
// Postgres jsonb 不保留键序(按键长+字节序重排),库里读出来的对象键序 ≠ 代码构造顺序。
// 不做规范化的话每次比较都判"变了" → 每日重算全量重写(本地实测连跑三次每次都刷新)。
test('顶层键序不同 → 不刷新', () => {
const asWritten = {
subKey: 'missing_tooth', triggers: [{ type: 'diagnosis', code: 'K08' }],
toothPosition: '27', daysSince: 119, signalOccurredAt: '2026-03-27T00:00:00.000Z',
expectedCategories: ['implant', 'prosthodontic'], focusCategory: 'prosthodontic', patientAge: 90,
};
// 库里 jsonb 实际存回来的顺序(生产实测形态)
const asStoredByJsonb = {
subKey: 'missing_tooth', triggers: [{ code: 'K08', type: 'diagnosis' }],
daysSince: 119, patientAge: 90, focusCategory: 'prosthodontic', toothPosition: '27',
signalOccurredAt: '2026-03-27T00:00:00.000Z', expectedCategories: ['implant', 'prosthodontic'],
};
expect(reasonNeedsRefresh(row(asStoredByJsonb), row(asWritten))).toBe(false);
});
test('嵌套对象(triggers 元素)键序不同 → 不刷新', () => {
const a = { ...BASE, triggers: [{ type: 'diagnosis', code: 'K08' }] };
const b = { ...BASE, triggers: [{ code: 'K08', type: 'diagnosis' }] };
expect(reasonNeedsRefresh(row(a), row(b))).toBe(false);
});
test('数组**顺序**仍然算变化(expectedCategories 是有序的:首项=主推)', () => {
const a = { ...BASE, expectedCategories: ['implant', 'prosthodontic'] };
const b = { ...BASE, expectedCategories: ['prosthodontic', 'implant'] };
expect(reasonNeedsRefresh(row(a), row(b))).toBe(true);
});
});
describe('红线:刷新判据不依赖文案格式', () => {
test('⭐ signals 相同 → 判不刷新,与 reason 文案无关', () => {
// reason 文案由 signals 派生(诊断名←triggers、牙位←toothPosition、
// 治疗类目←expectedCategories+patientAge),所以比 signals 就够;
// 不解析「N 天前」这类文本 —— 将来 scenario 改文案模板不会引发误刷。
expect(reasonNeedsRefresh(row({ ...BASE }), row({ ...BASE }))).toBe(false);
});
});
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment