Skip to content

P
pac
Commit 33a826e7 by luoqi
Options

feat(push): 存量/增量同一通道 —— 归一统一 / 乱序裁决 / stub 自建 / 并发闸 + 断流告警 

把 push(形态 A/C)打磨成宿主唯一接入通道,存量=按批重放的增量:

摄入一致性(消除同源分叉):
- 形态 C 复用 normalizeCanonical(金额元→分 / 无时区时间补 offset),含 envelope
  occurredAt/updatedAt —— 与形态 A/pull 同一转,宿主用自己系统的格式发,不做转换。
- 业务事件先于患者主档到达 → dispatcher stub 自建空壳患者(对齐 form A),主档后到补全
  同一行,不双人;push 跨批顺序无关。

幂等/乱序:
- patient_facts 加 source_updated_at,fact-writer 按它裁决:更旧版本晚到 → stale_skipped
  跳过,不旧覆新。关掉"补推快照 vs 实时推送并发交错"的乱序窗口。

体制/性能:
- 限批 500(A 从 2000 对齐 C);persona 入队去抖(5min 桶去重,存量批次合并重算)。
- 同 host 并发闸(信号量,PAC_PUSH_CONCURRENCY 默认 4;幂等+裁决保证并发语义=串行)。
- 单飞锁索引收窄 WHERE direction='pull' —— 原索引把并发 push 批次拦成 P2002/500。

异常/可观测:
- 整批全失败 → 拒收 30802 + SyncLog FAILED;失败率/疑似改表 → 告警。
- 未知 source → 10001(非 90000,避免宿主对永久错误无限重试)。
- push 断流监控(超 26h 无 push 告警);form A 响应透出 mappingMisses/suspectFields 供自检。

文档 channel-push 按宿主视角重写(subjectType 单一口径、金额时间统一"原样发 PAC 归一")。

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
parent 984e05cc
Pipeline #3339 failed in 0 seconds
Showing with 431 additions and 74 deletions
apps/pac-service/prisma/migrations/20260703005409_add_patient_facts_source_updated_at/migration.sql 0 → 100644
-- AlterTable
ALTER TABLE "patient_facts" ADD COLUMN "source_updated_at" TIMESTAMPTZ(3);
apps/pac-service/prisma/migrations/20260703021623_sync_lock_pull_only/migration.sql 0 → 100644
-- 单飞锁收窄到 pull 方向:
-- 锁的目的是串行化 pull / cold-import(游标互斥,一 host 一次只能跑一个)。
-- push(形态 A/C)每个批次也建一条 status='running' 的 sync_log,原索引把并发 push 批次
-- 一并拦成 P2002 → 500(push 的并发由 PushController 信号量管理,不需要这把锁)。
DROP INDEX "sync_logs_one_running_per_host";
CREATE UNIQUE INDEX "sync_logs_one_running_per_host"
ON "sync_logs" ("host_id")
WHERE "status" = 'running' AND "direction" = 'pull';
apps/pac-service/prisma/schema.prisma
......@@ -596,6 +596,10 @@ model PatientFact {
/// actual 应填 — 事实在宿主侧实际发生时间
occurredAt DateTime? @map("occurred_at") @db.Timestamptz(3)
/// 源记录末次修改时间(= canonicalRow.updatedAt,宿主侧口径)。乱序防护用:
/// 携带更旧 updatedAt 的版本晚到(补推快照 vs 实时推送并发)→ fact-writer 按此裁决跳过,不旧覆新。
/// null = 源没给 updatedAt → 退回"到达顺序"语义(向后兼容,老数据/无该列的 host 不受影响)。
sourceUpdatedAt DateTime? @map("source_updated_at") @db.Timestamptz(3)
/// planned 推荐填 — 未来安排的目标时间(预约时间 / 复查时间 / 计划治疗时间)
plannedFor DateTime? @map("planned_for") @db.Timestamptz(3)
/// planned 时间窗起;部分场景(医嘱"1-2 周内复查")需要区间
......@@ -1512,8 +1516,9 @@ model SyncLog {
@@index([status])
/// 单患者 pull 反查:"该患者最近被刷新过几次"
@@index([patientId, startedAt])
/// MIGRATION 20260528000000:partial UNIQUE (host_id) WHERE status='running' —
/// 同 host 同时只能有 1 个 running 行,作为存量/增量 sync 的并发锁。
/// MIGRATION 20260528000000 + 20260703021623:partial UNIQUE (host_id)
/// WHERE status='running' AND direction='pull' — 同 host 同时只能有 1 个 running 的
/// **pull** 行(存量/增量 sync 单飞锁);push 批次不受此锁(并发由 PushController 信号量管)。
/// Prisma 不支持声明式 partial UNIQUE,migration.sql 是单一真理源。
@@map("sync_logs")
}
apps/pac-service/src/modules/sync/assembler/field-mapper.ts
......@@ -223,7 +223,7 @@ function coerceArray(v: unknown): unknown[] | null {
* - 仅日期 'YYYY-MM-DD' → 直通(给 birthDate 用)
* - 无 tz('2026-05-10 14:00:00' / '2026-05-10T14:00:00')→ 按 timezone 配置补 offset
*/
function normalizeDatetime(s: string, timezone: string): string {
export function normalizeDatetime(s: string, timezone: string): string {
const trimmed = s.trim();
// 已有 tz
if (/[Zz]$/.test(trimmed) || /[+\-]\d{2}:?\d{2}$/.test(trimmed)) {
......
apps/pac-service/src/modules/sync/cold-import/cold-import.service.ts
......@@ -4,8 +4,9 @@ import * as path from 'node:path';
import * as yaml from 'js-yaml';
import { parse as parseCsv } from 'csv-parse/sync';
import { randomUUID } from 'node:crypto';
import { SyncDirection, SyncStatus } from '@pac/types';
import { ApiCode, SyncDirection, SyncStatus } from '@pac/types';
import type { Action, CanonicalResourceKey } from '@pac/types';
import { BizError } from '../../../common/errors/biz-error';
import type { Prisma } from '@prisma/client';
import { PrismaService } from '../../../prisma/prisma.service';
import { TransactionSynthesizer } from '../pipeline/transaction-synthesizer';
......@@ -286,6 +287,9 @@ export class ColdImportService {
syncLogId: string;
perResource: PerResourceStats[];
touched: Array<{ patientId: string; tenantId: string }>;
/** 映射覆盖缺口(落 _default 的原值种类)+ 疑似宿主改表 —— push 响应透出给宿主自检 */
mappingMisses: MappingMiss[];
suspectFields: SuspectField[];
}> {
const absDir = this.resolveHostDir(opts.hostName);
const manifest = this.readManifest(absDir);
......@@ -305,8 +309,11 @@ export class ColdImportService {
const known = [
...new Set(subjectCfgs.map((c) => traceRawSourceTable(c.primary.table, transforms))),
];
throw new Error(
`ingestRawTables: source='${opts.source}' 无匹配 assembler(可选源表=${known.join(',')})`,
// 契约码 10001(而非裸 Error→90000):这是宿主填错 source 的永久性错误,
// 按文档"9xxxx 可重试"的规则裸抛会让宿主对配置错误无限重试。
throw new BizError(
ApiCode.CLIENT_BAD_REQUEST,
`source='${opts.source}' 无匹配 assembler(可选源表=${known.join(',')})`,
);
}
......@@ -418,32 +425,74 @@ export class ColdImportService {
}
: null;
// 全批失败(非重复、无一落库)按 FAILED 算 —— 典型是宿主改了关键列(身份/时间)导致逐行合成失败,
// 旧口径只看 firstError 会把这种批标成 SUCCESS,宿主毫无感知地持续丢数据。
// 顺带收紧:部分行失败(failed>0)从 SUCCESS 改为 PARTIAL,让状态如实。
const allRowsFailed = agg.failed > 0 && agg.txn === 0 && agg.dup === 0;
const status =
(firstError && agg.txn === 0) || allRowsFailed
? SyncStatus.FAILED
: firstError || agg.failed > 0
? SyncStatus.PARTIAL
: SyncStatus.SUCCESS;
await this.prisma.syncLog.update({
where: { id: syncLog.id },
data: {
status:
firstError && agg.txn === 0
? SyncStatus.FAILED
: firstError
? SyncStatus.PARTIAL
: SyncStatus.SUCCESS,
status,
tenantId: [...seenTenants][0] ?? 't_unknown',
fetched: opts.rows.length,
transactionsWritten: agg.txn,
factsEmitted: agg.facts,
duplicates: agg.dup,
failed: agg.failed,
errorMessage: firstError,
errorMessage: firstError ?? (allRowsFailed ? '整批全部行失败(疑似关键字段缺失/改名)' : null),
endedAt: new Date(),
...(driftMeta ? { metadata: driftMeta as unknown as Prisma.InputJsonValue } : {}),
},
});
// ── push 路径主动告警(纯 push 宿主没有每日 pull cron 的探针,这里是唯一的"有人知道"渠道)──
// 只挑两类硬信号:疑似改表 / 失败率超阈。mappingMisses 不告警(长尾常态,靠 admin 页巡检 + 响应透出)。
const failedRatio = opts.rows.length > 0 ? agg.failed / opts.rows.length : 0;
if (suspectFields.length > 0) {
await this.alerter.send({
level: 'warning',
title: `push 疑似宿主改表:${opts.hostName}/${opts.source}`,
body: suspectFields.map((s) => `[${s.reason}] ${s.resource}.${s.field}`).join('\n'),
context: { host: opts.hostName, syncLogId: syncLog.id },
});
}
if (agg.failed > 0 && failedRatio >= 0.1) {
await this.alerter.send({
level: allRowsFailed ? 'critical' : 'warning',
title: `push 摄入失败率 ${(failedRatio * 100).toFixed(0)}%:${opts.hostName}/${opts.source}`,
body:
`rows=${opts.rows.length} txn=${agg.txn} dup=${agg.dup} failed=${agg.failed}` +
(firstError ? `\nfirstError: ${firstError}` : ''),
context: { host: opts.hostName, syncLogId: syncLog.id },
});
}
this.logger.log(
`ingestRawTables host=${opts.hostName} source=${opts.source} rows=${opts.rows.length} ` +
`txn=${agg.txn} dup=${agg.dup} failed=${agg.failed} touched=${touched.length}`,
);
return { syncLogId: syncLog.id, perResource, touched };
return { syncLogId: syncLog.id, perResource, touched, mappingMisses, suspectFields };
}
/**
* host 的归一化上下文(amountUnit/timezone)—— 给 push 形态 C 复用 Stage② normalizeCanonical,
* 让 A/C 的金额/时间语义严格同一(消除"C 跳过归一层"的分叉)。
* 无 manifest(纯 C 宿主未建配置目录)→ null,调用方用默认(fen + Asia/Shanghai)。
*/
getNormalizeContext(hostName: string): { amountUnit: 'yuan' | 'fen'; timezone: string } | null {
try {
const manifest = this.readManifest(this.resolveHostDir(hostName));
return { amountUnit: manifest.amount_unit, timezone: manifest.timezone };
} catch {
return null;
}
}
/// host → 配置目录(= sync.service.resolveDataDir 同约定:env PAC_INCREMENTAL_DATA_DIR ?? cwd/data,再拼 hostName)
......
apps/pac-service/src/modules/sync/pipeline/fact-writer.service.ts
......@@ -46,8 +46,10 @@ export class FactWriter {
sourceUnit: string;
patientId: string;
transactionId: string;
/// 源记录末次修改时间(canonicalRow.updatedAt)。乱序防护:更旧的版本晚到 → 跳过不旧覆新。
sourceUpdatedAt?: Date | null;
}): Promise<FactWriteResult> {
const { draft, hostId, tenantId, sourceUnit, patientId, transactionId } = input;
const { draft, hostId, tenantId, sourceUnit, patientId, transactionId, sourceUpdatedAt } = input;
// ─── 闸 1:写库前 zod 强校验(canonical-fact-layer.md §二)──
// 失败抛 FactContentSchemaError,由 ParserPipeline 计入 factsFailed 并 log issues。
......@@ -92,6 +94,25 @@ export class FactWriter {
const draftStatus = draft.status ?? FactStatus.ACTIVE;
// ── 乱序防护(业务时间裁决):双方都带 sourceUpdatedAt 且来的更旧 → 跳过,不旧覆新 ──
// 场景:补推批快照(T 时刻读)晚于实时推送到达。同值(同秒改两次)不拦,交 hash 决策。
// 任一方无 sourceUpdatedAt → 退回到达顺序语义(向后兼容)。
if (
latest?.sourceUpdatedAt &&
sourceUpdatedAt &&
sourceUpdatedAt.getTime() < latest.sourceUpdatedAt.getTime()
) {
this.logger.debug(
`stale skip subject=${draft.subjectId} incoming=${sourceUpdatedAt.toISOString()} < latest=${latest.sourceUpdatedAt.toISOString()}`,
);
return {
action: 'stale_skipped',
factId: latest.id,
subjectId: draft.subjectId,
version: latest.version,
};
}
if (latest) {
const latestHash = this.hashContent(latest.content as Prisma.InputJsonValue);
const sameContent = latestHash === newHash;
......@@ -146,6 +167,7 @@ export class FactWriter {
version: nextVersion,
clinicId: validatedDraft.clinicId ?? null,
occurredAt: validatedDraft.occurredAt ?? null,
sourceUpdatedAt: sourceUpdatedAt ?? null,
plannedFor: validatedDraft.plannedFor ?? null,
validFrom: validatedDraft.validFrom ?? null,
validUntil: validatedDraft.validUntil ?? null,
......@@ -235,6 +257,7 @@ export class FactWriter {
status: string;
hash: string;
transactionIds: string[];
sourceUpdatedAt: Date | null;
}>();
for (const [k, latest] of latestBySubject.entries()) {
liveLatest.set(k, {
......@@ -243,6 +266,7 @@ export class FactWriter {
status: latest.status,
hash: this.hashContent(latest.content as Prisma.InputJsonValue),
transactionIds: latest.transactionIds,
sourceUpdatedAt: latest.sourceUpdatedAt ?? null,
});
}
......@@ -252,6 +276,24 @@ export class FactWriter {
const live = liveLatest.get(vkey);
const draftStatus = entry.draft.status ?? FactStatus.ACTIVE;
// 乱序防护(与 writeDraft 同语义):更旧的版本晚到 → 跳过,不旧覆新
if (
live?.sourceUpdatedAt &&
entry.sourceUpdatedAt &&
entry.sourceUpdatedAt.getTime() < live.sourceUpdatedAt.getTime()
) {
this.logger.debug(
`stale skip(bulk) subject=${sid} incoming=${entry.sourceUpdatedAt.toISOString()} < latest=${live.sourceUpdatedAt.toISOString()}`,
);
results.push({
action: 'stale_skipped',
factId: live.id ?? '',
subjectId: sid,
version: live.version,
});
continue;
}
if (live && live.hash === entry.hash && live.status === draftStatus) {
// 内容一致 + 状态一致
if (live.id && !live.transactionIds.includes(entry.transactionId)) {
......@@ -293,6 +335,7 @@ export class FactWriter {
version: nextVersion,
clinicId: entry.draft.clinicId ?? null,
occurredAt: entry.draft.occurredAt ?? null,
sourceUpdatedAt: entry.sourceUpdatedAt ?? null,
plannedFor: entry.draft.plannedFor ?? null,
validFrom: entry.draft.validFrom ?? null,
validUntil: entry.draft.validUntil ?? null,
......@@ -310,6 +353,7 @@ export class FactWriter {
status: draftStatus,
hash: entry.hash,
transactionIds: [entry.transactionId],
sourceUpdatedAt: entry.sourceUpdatedAt ?? null,
});
results.push({
action: live ? 'superseded' : 'created',
......@@ -384,7 +428,8 @@ export interface FactWriteResult {
/// superseded — 旧 active 被置为 superseded,新版本插入
/// evidence_appended — 内容未变,但本次 transaction 被加进证据数组
/// unchanged — 内容未变且 transaction 已在证据数组中(完全幂等)
action: 'created' | 'superseded' | 'evidence_appended' | 'unchanged';
/// stale_skipped — 来的版本 sourceUpdatedAt 比库里更旧(乱序晚到)→ 忽略,不旧覆新
action: 'created' | 'superseded' | 'evidence_appended' | 'unchanged' | 'stale_skipped';
factId: string;
subjectId: string;
version: number;
......@@ -399,4 +444,6 @@ export interface BulkEntry {
sourceUnit: string;
patientId: string;
transactionId: string;
/// 源记录末次修改时间(canonicalRow.updatedAt);乱序防护裁决用,null = 源没给
sourceUpdatedAt?: Date | null;
}
apps/pac-service/src/modules/sync/pipeline/parser-pipeline.service.ts
......@@ -60,6 +60,7 @@ export class ParserPipeline {
factsSuperseded: 0,
factsUnchanged: 0,
factsEvidenceAppended: 0,
factsStaleSkipped: 0,
factsFailed: 0,
writes: [],
};
......@@ -79,6 +80,7 @@ export class ParserPipeline {
}
const drafts = parser.parse({ transaction, canonicalRow });
const sourceUpdatedAt = parseSourceUpdatedAt(canonicalRow);
// fact 的 source_unit = 该患者的 source_unit(集团内 subject_id 撞号消歧)
const patientSourceUnit =
......@@ -98,6 +100,7 @@ export class ParserPipeline {
sourceUnit: patientSourceUnit,
patientId: transaction.patientId,
transactionId: transaction.id,
sourceUpdatedAt,
});
metrics.writes.push(result);
switch (result.action) {
......@@ -113,6 +116,9 @@ export class ParserPipeline {
case 'evidence_appended':
metrics.factsEvidenceAppended++;
break;
case 'stale_skipped':
metrics.factsStaleSkipped++;
break;
}
} catch (err) {
metrics.factsFailed++;
......@@ -142,6 +148,7 @@ export class ParserPipeline {
factsSuperseded: 0,
factsUnchanged: 0,
factsEvidenceAppended: 0,
factsStaleSkipped: 0,
factsFailed: 0,
writes: [],
};
......@@ -171,6 +178,7 @@ export class ParserPipeline {
transaction: item.transaction,
canonicalRow: item.canonicalRow,
});
const sourceUpdatedAt = parseSourceUpdatedAt(item.canonicalRow);
for (const draft of drafts) {
bulkEntries.push({
draft,
......@@ -179,6 +187,7 @@ export class ParserPipeline {
sourceUnit: suMap.get(item.transaction.patientId) ?? '',
patientId: item.transaction.patientId,
transactionId: item.transaction.id,
sourceUpdatedAt,
});
}
} catch (err) {
......@@ -210,6 +219,9 @@ export class ParserPipeline {
case 'evidence_appended':
metrics.factsEvidenceAppended++;
break;
case 'stale_skipped':
metrics.factsStaleSkipped++;
break;
}
}
} catch (err) {
......@@ -226,6 +238,7 @@ export class ParserPipeline {
sourceUnit: e.sourceUnit,
patientId: e.patientId,
transactionId: e.transactionId,
sourceUpdatedAt: e.sourceUpdatedAt,
});
metrics.writes.push(r);
switch (r.action) {
......@@ -241,6 +254,9 @@ export class ParserPipeline {
case 'evidence_appended':
metrics.factsEvidenceAppended++;
break;
case 'stale_skipped':
metrics.factsStaleSkipped++;
break;
}
} catch (subErr) {
metrics.factsFailed++;
......@@ -278,6 +294,19 @@ export interface PipelineRunMetrics {
factsSuperseded: number;
factsUnchanged: number;
factsEvidenceAppended: number;
/// 乱序防护跳过数(更旧 sourceUpdatedAt 晚到,未旧覆新)
factsStaleSkipped: number;
factsFailed: number;
writes: FactWriteResult[];
}
/// canonicalRow.updatedAt(宿主末次修改时间,string / Date)→ Date;缺失或不可解析 → null(退回到达顺序语义)。
function parseSourceUpdatedAt(canonicalRow: Record<string, unknown>): Date | null {
const v = canonicalRow.updatedAt;
if (v instanceof Date) return isNaN(v.getTime()) ? null : v;
if (typeof v === 'string' && v.trim()) {
const d = new Date(v);
return isNaN(d.getTime()) ? null : d;
}
return null;
}
apps/pac-service/src/modules/sync/pipeline/pipeline-dispatcher.service.ts
......@@ -81,6 +81,8 @@ export class PipelineDispatcher {
* - patient 主档:返回 undefined(走 upsert 不进 transaction)
*/
emitsResolver: EmitsResolver;
/** persona 入队去抖(push 摄入路径传 true;语义见 QueueProducer.enqueuePersonaRecompute) */
personaDebounce?: boolean;
}): Promise<DispatchMetrics> {
const m = zeroMetrics();
/** 本批触碰过的 patientId 集合,batch 末统一 enqueue persona-recompute(去重) */
......@@ -131,6 +133,26 @@ export class PipelineDispatcher {
const rawRow = explicitSource ?? canonicalRow;
const sourceUnit =
explicitSourceUnit(canonicalRow) ?? resolveSourceUnit(rawRow, input.identityNamespaceField);
// stub auto-create(对齐 cold-import/form A 的 ensurePatientStub 语义):
// 业务事件先于患者主档到达 → 先建空壳患者(隔离键齐全),facts 正常衍生;
// 主档后到 upsert 命中同一行补全姓名电话,不双人 → push 跨批顺序无关。
// 旧行为是 patientId=null 落 txn + 跳过 parser"等下次 pull 回放"——纯 push 宿主没有 pull,数据会永久悬空。
const patientExternalId = canonicalRow.patientExternalId as string | undefined;
if (patientExternalId && !patientIndex.has(patientIndexKey(sourceUnit, patientExternalId))) {
try {
const stubId = await this.ensurePatientStub(
input.hostId,
input.tenantId,
sourceUnit,
patientExternalId,
);
patientIndex.set(patientIndexKey(sourceUnit, patientExternalId), stubId);
} catch (err) {
this.logger.warn(
`patient stub 自建失败 externalId=${patientExternalId}: ${err instanceof Error ? err.message : err}`,
);
}
}
const txn = this.synthesizer.synthesize({
rawRow,
canonicalRow,
......@@ -186,14 +208,17 @@ export class PipelineDispatcher {
m.personaEnqueued = touchedPatients.size;
for (const patientId of touchedPatients) {
try {
await this.queue.enqueuePersonaRecompute({
hostId: input.hostId,
tenantId: input.tenantId,
patientId,
triggeredBy: input.sourceContext.startsWith('txn:')
? input.sourceContext
: `txn:${input.sourceContext}`,
});
await this.queue.enqueuePersonaRecompute(
{
hostId: input.hostId,
tenantId: input.tenantId,
patientId,
triggeredBy: input.sourceContext.startsWith('txn:')
? input.sourceContext
: `txn:${input.sourceContext}`,
},
{ debounce: input.personaDebounce },
);
} catch (err) {
// 入队失败不破坏主流程,只 warn(凌晨 cron 兜底会扫到 stale persona)
this.logger.warn(
......@@ -211,10 +236,11 @@ export class PipelineDispatcher {
canonicalRow: Record<string, unknown>,
m: DispatchMetrics,
): Promise<void> {
// patientId 可能 null(push 首见 stub 患者尚未补全) → 跳过 parser,等下次 pull 补全后回放
// patientId=null 仅剩一种情形:事件本身没带患者引用(patientExternalId 缺失)——
// 带引用的未知患者已由 stub auto-create 兜住(见 dispatchTables)。无患者可挂 → 跳过 parser。
if (!txn.patientId) {
this.logger.debug(
`skip parser (patientId=null) subject=${txn.subjectId} action=${txn.action}`,
`skip parser (patientId=null,事件未带患者引用) subject=${txn.subjectId} action=${txn.action}`,
);
return;
}
......@@ -238,6 +264,30 @@ export class PipelineDispatcher {
m.factsFailed += r.factsFailed;
}
/// 空壳患者自建 —— 与 cold-import.ensurePatientStub 同语义(那边是 pull/form A 路径的原版):
/// 只填隔离键 + active,姓名等留空;真主档后到走 upsertPatient 命中同一行补全。
/// profile 也建空行(scenario SQL JOIN 需要 doNotContact/deceased 默认值)。
private async ensurePatientStub(
hostId: string,
tenantId: string,
sourceUnit: string,
externalId: string,
): Promise<string> {
const patient = await this.prisma.patient.upsert({
where: {
hostId_tenantId_sourceUnit_externalId: { hostId, tenantId, sourceUnit, externalId },
},
create: { hostId, tenantId, sourceUnit, externalId, active: true },
update: {}, // 已存在(含真主档)→ noop
});
await this.prisma.patientProfile.upsert({
where: { patientId: patient.id },
create: { patientId: patient.id, doNotContact: false, deceased: false, tags: [] },
update: {},
});
return patient.id;
}
private async upsertPatient(
row: Record<string, unknown>,
hostId: string,
......
apps/pac-service/src/modules/sync/push/push-receiver.service.ts
import { BadRequestException, Injectable, Logger } from '@nestjs/common';
import { SyncDirection, SyncStatus } from '@pac/types';
import { ApiCode, CanonicalResourceMeta, SyncDirection, SyncStatus } from '@pac/types';
import type { CanonicalResourceKey } from '@pac/types';
import { BizError } from '../../../common/errors/biz-error';
import { PrismaService } from '../../../prisma/prisma.service';
import {
normalizeCanonical,
normalizeDatetime,
type NormalizeContext,
} from '../assembler/field-mapper';
import { PipelineDispatcher, type EmitsResolver } from '../pipeline/pipeline-dispatcher.service';
import { ColdImportService } from '../cold-import/cold-import.service';
import { QueueProducer } from '../../../queues/queue-producer.service';
......@@ -52,17 +59,35 @@ export class PushReceiverService {
triggeredBy: `push:${hostName}:${source}`,
});
// 事件驱动:touched 患者入队 persona-recompute(BullMQ jobId 去重 + eventWatermark 幂等 + 凌晨 cron 兜底)
// ⚠️ plan 不在此触发(交给现有定时任务),与单患者刷新路径有意区分。
const agg = r.perResource.reduce(
(a, s) => ({
txn: a.txn + s.transactionsWritten,
dup: a.dup + s.duplicates,
failed: a.failed + s.failed,
}),
{ txn: 0, dup: 0, failed: 0 },
);
// ── 关键字段守卫:整批全失败(非重复、无一落库)→ 拒收报错,别让宿主静默丢数据 ──
// 典型成因:宿主改了身份/时间列名 → 逐行合成失败。失败行不落任何库(raw 只活在 txn),
// 拒绝是无损的;宿主修好后整批重推即可(幂等)。SyncLog 已标 FAILED + 告警(ingestRawTables 内)。
if (rows.length > 0 && agg.txn === 0 && agg.dup === 0 && agg.failed > 0) {
throw new BizError(
ApiCode.SYNC_CONTRACT_DRIFT,
`整批 ${agg.failed}/${rows.length} 行全部失败 —— 疑似关键字段缺失/改名,请核对数据字典` +
`(syncLogId=${r.syncLogId})`,
);
}
// 事件驱动:touched 患者入队 persona-recompute(去抖:存量批次重放时窗内合并,见 QueueProducer)
// ⚠️ plan 不在此触发(persona 完成后链式 enqueue),与单患者刷新路径有意区分。
let personaEnqueued = 0;
for (const { patientId, tenantId } of r.touched) {
try {
await this.queue.enqueuePersonaRecompute({
hostId,
tenantId,
patientId,
triggeredBy: `push:${hostName}`,
});
await this.queue.enqueuePersonaRecompute(
{ hostId, tenantId, patientId, triggeredBy: `push:${hostName}` },
{ debounce: true },
);
personaEnqueued++;
} catch (err) {
this.logger.warn(
......@@ -71,15 +96,6 @@ export class PushReceiverService {
}
}
const agg = r.perResource.reduce(
(a, s) => ({
txn: a.txn + s.transactionsWritten,
dup: a.dup + s.duplicates,
failed: a.failed + s.failed,
}),
{ txn: 0, dup: 0, failed: 0 },
);
this.logger.log(
`push/rows host=${hostName} source=${source} rows=${rows.length} ` +
`txn=${agg.txn} dup=${agg.dup} failed=${agg.failed} personaEnqueued=${personaEnqueued}`,
......@@ -93,6 +109,8 @@ export class PushReceiverService {
duplicates: agg.dup,
failed: agg.failed,
personaEnqueued,
mappingMisses: r.mappingMisses.length,
suspectFields: r.suspectFields.length,
};
}
......@@ -113,8 +131,8 @@ export class PushReceiverService {
// host→tenant 一致性校验:tenantId 必须是该 host 已建立的集团之一。
// 注:跨 host 隔离已由所有键的 host_id 保证(host 写不进别人数据);此处只防 host 自己
// 填错 tenant slug 把数据散进新分区。新 host(尚无任何数据)首次放行,由本次写入建立其集团;
// 接新集团请先经 cold-import / pull 建立,push 再跟上。
// 填错 tenant slug 把数据散进新分区。新 host(尚无任何数据)首次放行,由本次写入建立其集团
// —— 存量可全程走 push(按批串行重放,见 channel-push 文档"存量接入"),不再要求先 cold-import。
const knownTenants = await this.prisma.patient.groupBy({
by: ['tenantId'],
where: { hostId },
......@@ -140,8 +158,15 @@ export class PushReceiverService {
},
});
// ─── 2. events → tables(按 subjectType 分组,canonicalRow 自带元字段 _action/_subjectType) ───
const tables = this.eventsToTables(events);
// ─── 2. events → tables(按 subjectType 分组 + Stage② 归一)───
// 归一与形态 A / pull 走同一个 normalizeCanonical(消除同源分叉):金额(元→分)/
// 无时区时间(补 offset)/ 布尔 coerce,按全局 CanonicalResourceMeta 识别列。
// 配置来自 manifest(纯 C 宿主没建 → 默认 分 + Asia/Shanghai,即原契约语义,向后兼容)。
const normCtx: NormalizeContext = this.coldImport.getNormalizeContext(hostName) ?? {
amountUnit: 'fen',
timezone: 'Asia/Shanghai',
};
const tables = this.eventsToTables(events, normCtx);
// ─── 3. dispatch(emits resolver:per-row,从行的 _action / _subjectType 元字段读) ───
let metrics;
......@@ -154,6 +179,7 @@ export class PushReceiverService {
sourceContext: `push:${hostName}`,
syncLogId: syncLog.id,
emitsResolver: this.buildEmitsResolver(),
personaDebounce: true, // push 摄入统一去抖(存量批次重放时窗内合并)
});
} catch (err) {
firstError = err instanceof Error ? err.message : String(err);
......@@ -206,12 +232,14 @@ export class PushReceiverService {
}
/**
* 把 events[] 转换成 PipelineDispatcher 需要的 tables 形态。
* 把 events[] 转换成 PipelineDispatcher 需要的 tables 形态,并做 Stage② 归一。
*
* 约定:event.subjectType 决定 table key(多元化:patient → patients,appointment → appointments)。
* 每个 canonical row 由 event.payload + 元字段(externalId / clinicId / occurredAt 等)合成。
* 归一按**单数** subjectType 查 CanonicalResourceMeta(表 key 是复数,meta key 是单数,别搞混);
* 未知 subjectType 不归一,交后续管线处理。
*/
private eventsToTables(events: PushEvent[]): Record<string, unknown[]> {
private eventsToTables(events: PushEvent[], normCtx: NormalizeContext): Record<string, unknown[]> {
const tables: Record<string, unknown[]> = {};
for (const e of events) {
......@@ -219,22 +247,31 @@ export class PushReceiverService {
if (!tables[tableKey]) tables[tableKey] = [];
// canonical row = payload + 元字段拍平
const canonicalRow: Record<string, unknown> = {
// envelope 时间字段(occurredAt/updatedAt)与 payload 同一转:无时区按配置补 offset
//(synthesizer 的 new Date() / 乱序裁决的 parseSourceUpdatedAt 才不会按服务器时区误解)
let canonicalRow: Record<string, unknown> = {
...e.payload,
externalId: e.subjectId,
clinicId: e.clinicId,
occurredAt: e.occurredAt,
occurredAt: normalizeDatetime(e.occurredAt, normCtx.timezone),
// 源组织命名空间来自事件信封 → dispatcher 优先取它(防跨命名空间撞号)
sourceUnit: e.sourceUnit ?? '',
...(e.patientId ? { patientExternalId: e.patientId } : {}),
// 末次修改时间 → synthesizer 合幂等键(内容变升版本)
...(e.updatedAt ? { updatedAt: e.updatedAt } : {}),
// 末次修改时间 → synthesizer 合幂等键(内容变升版本)+ 乱序裁决
...(e.updatedAt ? { updatedAt: normalizeDatetime(e.updatedAt, normCtx.timezone) } : {}),
// 给 synthesizer / dispatcher 用的元字段
_action: e.action,
_subjectType: e.subjectType,
// ⭐ 真原文留底:host event payload(不含 PAC 加的 meta 字段)
__source_row: e.payload,
};
if (e.subjectType in CanonicalResourceMeta) {
canonicalRow = normalizeCanonical(
canonicalRow,
e.subjectType as CanonicalResourceKey,
normCtx,
);
}
tables[tableKey]!.push(canonicalRow);
}
......
apps/pac-service/src/modules/sync/push/push.controller.ts
......@@ -8,6 +8,8 @@ import {
} from '@nestjs/common';
import type { Request } from 'express';
import { ApiOperation, ApiTags } from '@nestjs/swagger';
import { ApiCode } from '@pac/types';
import { BizError } from '../../../common/errors/biz-error';
import { Public } from '../../../common/decorators/public.decorator';
import { HmacVerifier } from './hmac-verifier.service';
import { PushReceiverService } from './push-receiver.service';
......@@ -44,6 +46,49 @@ export class PushController {
private readonly receiver: PushReceiverService,
) {}
/**
* 同 host 摄入并发闸(/events + /rows 共享)—— 磨平存量/增量的体制约束之一:
* push 在服务进程内同步跑管线,存量重放是成百上千个批次连续打;并发闸 + 限批
* = 天然限流,不与线上用户请求抢连接池。
*
* 额度 PAC_PUSH_CONCURRENCY(默认 4)。幂等 + updatedAt 乱序裁决保证并发与串行语义
* 完全一致(宿主串行推也行,只是慢);额度守住连接池(4 批 × 批内串行写 ≈ 数个连接)。
* 契约:超额并发返 CLIENT_RATE_LIMITED(10003),宿主退避重发即可。
* 单进程内存计数即可(服务单实例部署;将来多实例再升 Redis 信号量)。
*/
private readonly inflightByHost = new Map<string, number>();
/** 手动 zod parse → 契约码 10002(裸 ZodError 会被兜成 90000,与文档"字段校验失败=10002"不符) */
private parseOrReject<T>(parse: () => T): T {
try {
return parse();
} catch (err) {
throw new BizError(
ApiCode.CLIENT_VALIDATION_FAILED,
`push body 校验失败:${err instanceof Error ? err.message.slice(0, 300) : String(err)}`,
);
}
}
private async withHostLock<T>(hostId: string, hostName: string, fn: () => Promise<T>): Promise<T> {
const limit = Math.max(1, Number(process.env.PAC_PUSH_CONCURRENCY) || 4);
const cur = this.inflightByHost.get(hostId) ?? 0;
if (cur >= limit) {
throw new BizError(
ApiCode.CLIENT_RATE_LIMITED,
`host=${hostName} 并发批次已达上限(${limit});请退避后重发本批`,
);
}
this.inflightByHost.set(hostId, cur + 1);
try {
return await fn();
} finally {
const n = (this.inflightByHost.get(hostId) ?? 1) - 1;
if (n <= 0) this.inflightByHost.delete(hostId);
else this.inflightByHost.set(hostId, n);
}
}
@Post('events')
@Public()
@HttpCode(200)
......@@ -68,15 +113,17 @@ export class PushController {
rawBody,
});
// body 单独走 zod 校验(Nest 全局 ZodValidationPipe 已挂)
// body 单独走 zod 校验(手动 parse → 契约码 10002)
// 集团 tenantId 由事件携带(receiver 内解析校验),不再由 host 名兜底。
const parsed = PushEventsRequestSchema.parse(body);
const parsed = this.parseOrReject(() => PushEventsRequestSchema.parse(body));
return this.receiver.receive({
hostId,
hostName,
events: parsed.events,
});
return this.withHostLock(hostId, hostName, () =>
this.receiver.receive({
hostId,
hostName,
events: parsed.events,
}),
);
}
/**
......@@ -109,13 +156,15 @@ export class PushController {
rawBody,
});
const parsed = PushRowsRequestSchema.parse(body);
const parsed = this.parseOrReject(() => PushRowsRequestSchema.parse(body));
return this.receiver.receiveRows({
hostId,
hostName,
source: parsed.source,
rows: parsed.rows,
});
return this.withHostLock(hostId, hostName, () =>
this.receiver.receiveRows({
hostId,
hostName,
source: parsed.source,
rows: parsed.rows,
}),
);
}
}
apps/pac-service/src/modules/sync/push/push.schema.ts
......@@ -26,9 +26,11 @@ export const PushEventSchema = z.object({
/// 源组织命名空间 id;集团内给患者号消歧,进幂等键防跨命名空间撞号。单一命名空间可省(='')。
sourceUnit: z.string().optional().default('').describe('源组织命名空间 id'),
clinicId: z.string().min(1).describe('发生诊所(操作类必填)'),
occurredAt: z.string().datetime({ offset: true }).describe('ISO 8601 with timezone'),
/// 末次修改时间;进幂等键 → 内容变升版本、重试不变去重。缺省回退 occurredAt。
updatedAt: z.string().optional().describe('末次修改时间;驱动幂等 / 版本'),
/// 时间格式随宿主系统(带时区直通;无时区按接入配置的 timezone 归一)—— 与形态 A 同一转,
/// 宿主无需为 PAC 做格式转换(口径统一,消除"envelope 例外"心智负担)。
occurredAt: z.string().min(1).describe('发生时间;无时区时按接入配置解释'),
/// 末次修改时间;进幂等键 → 内容变升版本、重试不变去重;乱序裁决用。缺省回退 occurredAt。
updatedAt: z.string().optional().describe('末次修改时间;驱动幂等 / 版本 / 乱序裁决'),
eventSeq: z.number().int().nonnegative().optional().describe('递增序号,可选,用于水位'),
payload: z.record(z.string(), z.unknown()).describe('已 canonical 化的业务字段'),
});
......@@ -69,8 +71,10 @@ export const PushRowsRequestSchema = z.object({
rows: z
.array(z.record(z.string(), z.unknown()))
.min(1)
.max(2000)
.describe('原生表行,字段照抄'),
// 500 上限 = 磨平存量/增量的体制约束:单批处理时间可控(webhook 同步跑管线不超时),
// 存量 = 同一通道按批串行重放,与形态 C 的 events 上限对齐。
.max(500)
.describe('原生表行,字段照抄;单批 ≤500,存量分批串行推'),
});
export type PushRowsRequest = z.infer<typeof PushRowsRequestSchema>;
......@@ -83,6 +87,9 @@ export const PushRowsResponseSchema = z.object({
duplicates: z.number().int(),
failed: z.number().int(),
personaEnqueued: z.number().int().describe('入队画像重算的患者数(plan 不在此触发,由定时任务保)'),
/// 宿主自检信号(代替 cold-import 的 dry-run):样本批推完看这两个数,>0 先停下联系 PAC。
mappingMisses: z.number().int().describe('映射覆盖缺口种类数(有原值落 _default)'),
suspectFields: z.number().int().describe('疑似宿主改表项数(列缺席/新增,vs 历史基线)'),
});
export type PushRowsResponse = z.infer<typeof PushRowsResponseSchema>;
apps/pac-service/src/queues/dw-lag-monitor.service.ts
......@@ -92,5 +92,43 @@ export class DwLagMonitorService {
this.logger.log(msg);
}
}
await this.checkPushLag(hosts, now);
}
/**
* push 断流检测(纯 push 宿主的唯一"数据还活着"探针)。
* 只对**曾经推过**的 host 生效(有 push SyncLog)—— 还没上线 push 的 host 不误报。
* 阈值 PAC_PUSH_LAG_ERROR_HOURS(默认 26h:容忍"每日一推/每晚补推"节奏 + 2h 余量)。
*/
private async checkPushLag(
hosts: Array<{ id: string; name: string }>,
now: number,
): Promise<void> {
const errorH = Number(process.env.PAC_PUSH_LAG_ERROR_HOURS ?? '26');
for (const host of hosts) {
const lastPush = await this.prisma.syncLog.findFirst({
where: { hostId: host.id, direction: 'push' },
orderBy: { startedAt: 'desc' },
select: { startedAt: true },
});
if (!lastPush) continue; // 从未推过 = 未上线 push,不监控
const diffHours = (now - lastPush.startedAt.getTime()) / 3_600_000;
if (diffHours > errorH) {
this.logger.error(
`push-lag: 🔴 host=${host.name} 已 ${diffHours.toFixed(1)}h 无 push(阈值 ${errorH}h)`,
);
await this.alerter.send({
level: 'critical',
title: 'push 断流:宿主推送中断',
body:
`host=${host.name} 最近一次 push 在 ${lastPush.startedAt.toISOString()},` +
`已 ${diffHours.toFixed(1)} 小时无推送(阈值 ${errorH}h)— 宿主侧推送链路可能挂了,数据在漏。`,
context: { host: host.name, lastPushAt: lastPush.startedAt.toISOString() },
});
} else {
this.logger.log(`push-lag: 🟢 host=${host.name} last_push=${diffHours.toFixed(1)}h ago`);
}
}
}
}
apps/pac-service/src/queues/queue-producer.service.ts
......@@ -31,7 +31,35 @@ export class QueueProducer {
@InjectQueue(QueueName.EXECUTION_CALLBACK) private readonly callbackQ: Queue,
) {}
async enqueuePersonaRecompute(job: PersonaRecomputeJob): Promise<void> {
async enqueuePersonaRecompute(
job: PersonaRecomputeJob,
opts?: {
/**
* 去抖模式(push 摄入路径用)—— 磨平存量/增量差异的关键:
* 存量按批重放时同一患者被几十个批次连续 touch,立即入队会重算几十遍(每遍读全量历史,
* Node-CPU-bound)。去抖 = delay 一个时间窗 + 窗内按 (patientId, 时间桶) 去重:
* 窗内多批合并成一次重算(job 到期跑时读的是当时 DB 全量,晚落的批自动包含);
* 增量只多付一个窗的延迟(召回是日级节奏,无感)。
* 非 push 路径(单患者刷新等要即时反馈)不传,行为不变。
*/
debounce?: boolean;
},
): Promise<void> {
if (opts?.debounce) {
const windowMs = resolvePersonaDebounceMs();
// 时间桶 jobId:同患者同桶内的重复 add 被 BullMQ jobId 去重吃掉,只留第一个(delayed);
// 跨桶产生新 job —— 最坏每患者每窗 1 次重算,50 批连续推 → 2~3 次而非 50 次。
const bucket = Math.floor(Date.now() / windowMs);
await this.personaQ.add(job.patientId, job, {
jobId: `persona_${job.patientId}_${bucket}`,
delay: windowMs,
...QueueDefaults[QueueName.PERSONA_RECOMPUTE],
});
this.logger.debug(
`enqueue persona-recompute(debounced ${windowMs}ms) patient=${job.patientId} triggeredBy=${job.triggeredBy}`,
);
return;
}
// jobId 用 _ 分隔(BullMQ 不允许 :);同 patient 的 waiting job 自动去重
await this.personaQ.add(job.patientId, job, {
jobId: `persona_${job.patientId}`,
......@@ -73,3 +101,11 @@ export class QueueProducer {
);
}
}
/// persona 去抖窗口(push 摄入路径):PAC_PUSH_PERSONA_DEBOUNCE_MS,默认 5 分钟。
/// 太短(<1min)合并不了存量批次;太长增量画像延迟大。下限 10s 防误配成 0 把去抖变空转。
function resolvePersonaDebounceMs(): number {
const raw = Number(process.env.PAC_PUSH_PERSONA_DEBOUNCE_MS);
if (Number.isFinite(raw) && raw >= 10_000) return raw;
return 5 * 60_000;
}
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