Skip to content

P
pac
Commit 120db745 by luoqi
Options

fix(meyer): isolate clinic checkpoints and bound shared polling

parent 5d6e909a
Pipeline #3844 failed in 0 seconds
Showing with 471 additions and 51 deletions
apps/pac-service/src/cli/meyer-imaging.cli.ts
......@@ -4,10 +4,11 @@ import { runMeyerTick } from '../integrations/meyer/scheduler';
import { readFile } from 'node:fs/promises';
import { resolve } from 'node:path';
import { NestFactory } from '@nestjs/core';
import { loadMeyerScopes } from '../integrations/meyer/org-mapping';
import { loadMeyerScopes, loadMeyerScopeResults } from '../integrations/meyer/org-mapping';
import { ConfigSchema, InputConfigSchema, hash, imageKey } from '../integrations/meyer/contracts';
import { MeyerClient } from '../integrations/meyer/client';
import { collectMeyer } from '../integrations/meyer/collector';
import { prepareClinicStates } from '../integrations/meyer/clinic-state';
import { backfillMeyerFileRefs } from '../integrations/meyer/backfill-file-refs';
import { PrismaClient } from '@prisma/client';
import { loadDiagnosisCodes } from '../integrations/meyer/mapper';
......@@ -34,7 +35,7 @@ export function parseMeyerArgs(argv: string[]) {
if (args.tick && !args.apply) throw new Error('--tick/--watch requires --apply; preview cannot advance cursor');
return args;
}
export async function runMeyerCli(argv: string[]): Promise<number> {
export async function runMeyerCli(argv: string[], isolateWatchFailures = false): Promise<number> {
const args = parseMeyerArgs(argv);
if (args.help) {
console.log('meyer-imaging --config=<json> --state-dir=<private-directory> [--apply] [--tick | --watch] [--image-files=<imageId>] [--backfill-file-refs]\n' +
......@@ -46,15 +47,18 @@ export async function runMeyerCli(argv: string[]): Promise<number> {
const parsed = InputConfigSchema.safeParse(JSON.parse(await readFile(resolve(args.config), 'utf8')));
if (!parsed.success) throw new Error('invalid Meyer config; check required fields, scope, dates and HTTPS base URL');
const input = parsed.data;
// watch 的每轮重新读取组织数据；映射变化会被既有 state identity 拦截。
// 每轮核对组织映射；单诊所异常隔离，已登记诊所的归属变化仍由私有状态拒绝。
const { clinicIds, ...fields } = input;
const config = ConfigSchema.parse({ ...fields, scopes: await loadMeyerScopes(clinicIds) });
const resolved = args.tick ? await loadMeyerScopeResults(clinicIds)
: { scopes: await loadMeyerScopes(clinicIds), failures: [] };
if (!resolved.scopes.length) throw new Error('Meyer clinic mapping missing, conflicting or invalid');
const config = ConfigSchema.parse({ ...fields, scopes: resolved.scopes });
if (args.watch) {
const controller = new AbortController();
const stop = () => controller.abort();
process.once('SIGINT', stop); process.once('SIGTERM', stop);
try {
await watchMeyer(config, () => withRoundDeadline(() => runMeyerCli([...argv.filter(a => a !== '--watch'), '--tick']), config.roundTimeoutMs), controller.signal);
await watchMeyer(config, () => withRoundDeadline(() => runMeyerCli([...argv.filter(a => a !== '--watch'), '--tick'], true), config.roundTimeoutMs), controller.signal);
return 0;
} finally { process.removeListener('SIGINT', stop); process.removeListener('SIGTERM', stop); }
}
......@@ -63,16 +67,22 @@ export async function runMeyerCli(argv: string[]): Promise<number> {
const client = new MeyerClient(config, { appKey, secret: process.env.MEYER_APP_SECRET ?? '' });
const codes = loadDiagnosisCodes(resolve(__dirname, '../../data/_shared/dict/diagnosis-kcode.yaml'));
const store = new MeyerStore(resolve(args.state));
const release = await store.acquire({ baseUrl: config.baseUrl.replace(/\/$/, ''), appKey,
namespace: config.namespace, host: config.host, timezone: config.timezone,
scopes: [...config.scopes].sort((a, b) => a.clinicId.localeCompare(b.clinicId)) });
const identity = { baseUrl: config.baseUrl.replace(/\/$/, ''), appKey,
namespace: config.namespace, host: config.host, timezone: config.timezone };
const dbTarget = new URL(process.env.DATABASE_URL!);
const target = hash([dbTarget.hostname, dbTarget.port, dbTarget.pathname, dbTarget.searchParams.get('schema') ?? 'public']);
const validateTarget = async () => {
const previous = await store.read<{ target: string }>('database');
if (previous && previous.target !== target) throw new Error('state directory belongs to another PAC database');
};
const release = await store.acquire({ version: 2, ...identity }, [{ ...identity,
scopes: [...config.scopes].sort((a, b) => a.clinicId.localeCompare(b.clinicId)) }],
async () => { await validateTarget(); await prepareClinicStates(store, config); });
let app: Awaited<ReturnType<typeof NestFactory.createApplicationContext>> | undefined;
try {
if (!process.env.DATABASE_URL) throw new Error('DATABASE_URL required for organization mapping');
const db = new URL(process.env.DATABASE_URL);
const target = hash([db.hostname, db.port, db.pathname, db.searchParams.get('schema') ?? 'public']);
await validateTarget();
const previous = await store.read<{ target: string }>('database');
if (previous && previous.target !== target) throw new Error('state directory belongs to another PAC database');
if (!previous) await store.write('database', { target });
if (args.backfillFileRefs) {
const db = new PrismaClient();
......@@ -110,12 +120,17 @@ export async function runMeyerCli(argv: string[]): Promise<number> {
}
if (args.tick) {
const status = await runMeyerTick(config, c => client.forWindow(c), store, codes, sink!);
console.log(JSON.stringify({ mode: 'tick', pending: status.pending, autoPending: status.autoPending,
await store.write('scheduler-status', { ...status, scopeFailures: resolved.failures });
console.log(JSON.stringify({ mode: 'tick', clinicFailures: status.clinicFailures,
mappingFailures: resolved.failures.length, sourceRequests: status.sourceRequests, clinicsVisited: status.clinics.length, pending: status.pending, autoPending: status.autoPending,
manualReview: status.manualReview, retiredUnsupported: status.retiredUnsupported, retryAttempted: status.retryAttempted,
retryFailed: status.retryFailed, scanned: status.scan?.examined ?? 0,
scanComplete: status.scan?.complete ?? false, scanFailed: status.scan?.failed ?? 0,
imagesApplied: status.scan?.imagesApplied ?? 0, analysisFailed: status.scan?.analysisFailed ?? 0 }));
return status.retryFailed || status.scan?.failed || status.scan?.analysisFailed ? 1 : status.pending ? 2 : 0;
const failed = resolved.failures.length || status.clinicFailures || status.queueErrors || status.retryFailed ||
status.scan?.failed || status.scan?.analysisFailed;
// 诊所故障已有独立退避；watch 不再因此降低正常诊所的调度频率。共享故障仍抛出。
return failed && !isolateWatchFailures ? 1 : status.pending ? 2 : 0;
}
const result = await collectMeyer(config, client, store, codes, sink);
// 控制台只输出数量；原文、患者标识和逐条错误留在私有状态目录。
......
apps/pac-service/src/integrations/meyer/README.md
......@@ -63,7 +63,7 @@ pnpm meyer-imaging -- --config=/absolute/meyer.json --state-dir=/absolute/privat
退出码：0=所扫范围无失败或待处理；1=请求/校验/写入失败；2=仍有待处理记录。0 不代表美亚全部历史数据或所有分析均已同步。终端只输出计数；细项见 `last-run.json`，原始响应见 `observed-*`，影像标准字段及应用状态见 `image-*`，分析快照见 `bundle-*`。
影像和分析分别在对应事实成功落库后标记。影像元数据与分析使用两个稳定事件身份，影像业务 ID 保持一致；分析到达后由原有 FactWriter 更新同一影像的版本，并关联牙位诊断。部分失败使用已留底快照恢复；重复执行不增加重复事实，重放元数据不会清空分析。状态目录有进程锁；明确属于本机已退出进程的锁可自动回收。存活进程、其他机器、未知 owner 或回收操作本身中断的锁不自动抢占。更换范围映射、应用或 PAC 数据库须使用独立目录；单独扩大时间窗可复用原目录。
影像和分析分别在对应事实成功落库后标记。影像元数据与分析使用两个稳定事件身份，影像业务 ID 保持一致；分析到达后由原有 FactWriter 更新同一影像的版本，并关联牙位诊断。部分失败使用已留底快照恢复；重复执行不增加重复事实，重放元数据不会清空分析。状态目录有进程锁；明确属于本机已退出进程的锁可自动回收。存活进程、其他机器、未知 owner 或回收操作本身中断的锁不自动抢占。更换应用或 PAC 数据库须使用独立目录。诊所名单可增删：新诊所从配置 timeStart 起独立扫描，移除诊所停止调度、保留原状态；原诊所租户/品牌变化仍拒绝续跑，不能靠改名单重置归属。单独扩大时间窗可复用原目录。
不要把补充记录交给 Friday 原表 YAML 的通用 reparse；本路线以自己的留底快照重试。首次 preview 留底后，apply 使用同一快照，不重新解释为最新结果。已应用文件不可直接改写或删除来模拟对账。
......@@ -85,18 +85,20 @@ pnpm meyer-imaging -- --config=/absolute/meyer.json --state-dir=/absolute/privat
pnpm meyer-imaging -- --config=/absolute/meyer.json --state-dir=/absolute/private/meyer-test --apply --watch
```
诊所范围是授权边界，不应因定时运行而扩大。首次从配置 timeStart 开始，后续使用私有 cursor。tick/watch 的结束时间由当前时间决定（固定窗口 timeEnd 仅用于默认单次采集），每轮最多推进 1 小时，落后当前时刻至少 1 分钟，前后窗口重叠 1 分钟。时间窗口按影像入库时间划分，不要求影像在窗口内已经分析；晚到分析由 pending 跨窗口追踪。时间窗口不是分析更新时间，不保证捕获任意历史修订。
诊所范围是授权边界，不应因定时运行而扩大。每家诊所首次从配置 timeStart 开始，后续使用各自的私有 cursor。tick/watch 的结束时间由当前时间决定（固定窗口 timeEnd 仅用于默认单次采集），每轮最多推进 1 小时，落后当前时刻至少 1 分钟，前后窗口重叠 1 分钟。时间窗口按影像入库时间划分，不要求影像在窗口内已经分析；晚到分析由 pending 跨窗口追踪。时间窗口不是分析更新时间，不保证捕获任意历史修订。
单次固定窗口用于联调或明确范围的历史补采，不会自动变成定时任务。持续采集须由外部守护真正运行 `--watch`，或定时运行 `--tick`；仅保存配置不会启用。历史跨度较大时应先完成明确范围的补采，随后使用同一目录从初始时间逐窗确认水位；不要把一次窄窗成功当成历史完整，也不要手改 cursor 跳过尚未扫描的区间。
- 新影像先经过原有标准解析/事实写入流程入库。没有分析、分析请求失败或分析写入失败时保留 pending/，允许影像扫描水位前进；分页、身份校验或影像落库失败时不推进。
- `imagesApplied` 记录本轮新写入影像阶段数，`applied` 记录完成分析阶段数；`failed` 是阻止影像扫描的失败，`analysisFailed` 单独记录分析请求/写入异常，两类均通过 CLI 失败退出码反映。空分析计入 deferred，不制造分析完成或阴性结论。
- 每轮先处理最早到期的最多 10 个（可配置）待处理项。全景片从影像入库时刻起，前 15 分钟按正常轮次补查，之后每小时最多查一次。累计实际调用分析接口 8 次仍未取得结果，停止自动查询，转人工处理；记录和已入库影像保留。旧队列的混合重试次数不计入分析次数。
- 一个定时器串行轮转诊所，默认每轮最多 5 家；进度、分页断点、待处理分组与故障退避按诊所隔离。某家失败或退避不阻塞其他诊所，轮转位置在请求前保存，重启也不会长期停留在队首。
- 重试 10 张、新影像扫描 100 条、扫描 10 页、列表/分析业务请求 100 次均为整轮共享上限，不乘诊所数；业务请求额度不含令牌获取及鉴权刷新。按本轮诊所数分配额度，未处理内容留待后续轮次。诊所多于轮次容量时，单家巡检间隔会增长，应根据积压和水位调整这组公共额度。
- 每家在分得的额度内，优先重试最早到期的待处理项。全景片从影像入库时刻起，前 15 分钟按正常轮次补查，之后每小时最多查一次。累计实际调用分析接口 8 次仍未取得结果，停止自动查询，转人工处理；记录和已入库影像保留。旧队列的混合重试次数不计入分析次数。
- 当前只支持全景片分析（aiType=3，兼容明确的全景片名称/类型）。其他类型只保存影像，不进入分析补查队列；类型不明时保留人工核实。分类记录写入 analysis-status，不能解释为分析阴性。升级时本地清理已落库且没有待写分析快照的旧非分析类型，不额外请求美亚、不占重试名额。
- 次数在请求前持久化，进程重启或重叠扫描不会重置。分析请求异常也计次数；已取得分析但写入失败的，使用原快照继续恢复写入，不受查询上限限制。身份校验、患者未绑定和影像写入失败另行退避，不消耗分析次数。
- 重试通过原影像入库时间前后 1 秒的窄窗重新定位 imageId，再获取分析。不会扫描整个诊所历史。身份改绑、影像不可见均保留待处理并报错，不自动撤销或迁移事实。
- 因接口没有按 imageId 直接查询元数据的能力，缺少入库时间的条目不能自动定位，须人工处理；这类情况会有 retryFailed。
- scheduler-status.json 提供扫描结果、水位、总待处理数 pending、自动补查数 autoPending、人工处理数 manualReview、当轮移出的不适用条数 retiredUnsupported 和重试失败数。watch 遇网络、美亚或数据库异常会记录失败计数并退避，默认从 15 分钟增长到最多 1 小时；依赖恢复后继续，不丢弃失败数据。配置错误、损坏状态或归属冲突需要修复后才能恢复。SIGINT/SIGTERM 会等当前一轮收尾再停止。
- scheduler-status.json 提供本轮各诊所结果、独立水位、sourceRequests、clinicFailures、组织映射异常 scopeFailures，以及总待处理数 pending、自动补查数 autoPending、人工处理数 manualReview、当轮移出的不适用条数 retiredUnsupported 和重试失败数。单家诊所故障按 15 分钟至最多 1 小时退避；单家组织映射缺失或冲突也单独记录，正常诊所继续。数据库不可用等共享依赖故障仍可能影响整轮；watch 仅对共享依赖异常额外退避，不因单诊所失败降低其他诊所频率；依赖恢复后继续，不丢弃失败数据。配置错误、损坏状态或归属冲突需要修复后才能恢复。SIGINT/SIGTERM 会等当前一轮收尾再停止。
- 已成功分析的影像不再重新查询分析；只有影像阶段完成且符合补查策略时才自动重试。这是首次接入及晚到分析重试，不是历史对账；删除、撤回和成功后的修改仍不覆盖。
- 状态目录必须持久化。先前仅有 deferred-* 的旧目录没有完整队列，升级后先按原窗口重跑一次再启动 tick/watch。不要用曾做远期手工采集的目录直接推断历史已覆盖。
......@@ -142,7 +144,9 @@ pnpm meyer-imaging -- --config=/absolute/meyer.json --state-dir=/absolute/privat
| roundTimeoutMs | 600000 | watch 单轮最多 10 分钟；超时退出由外部守护重启 |
| pollIntervalMs | 900000 | 一轮结束后等 15 分钟 |
| maxNewImagesPerTick | 100 | 所有配置诊所合计最多处理 100 条扫描记录；含重复/待处理项 |
| maxRetriesPerTick | 10 | 每轮最多重试 10 张待处理影像 |
| maxRetriesPerTick | 10 | 所有诊所合计每轮最多重试 10 张待处理影像 |
| maxClinicsPerTick | 5 | 每轮最多轮转诊所数，同时受公共扫描/请求额度约束 |
| maxSourceRequestsPerTick | 100 | 所有诊所合计列表/分析业务请求上限；不含令牌获取及鉴权刷新 |
| analysisSlowAfterMs | 900000 | 影像入库 15 分钟后转低频补查 |
| analysisSlowRetryMs | 3600000 | 低频补查间隔 1 小时，且不短于正常轮次间隔 |
| maxAnalysisAttempts | 8 | 每张影像最多自动查询分析 8 次，含首次查询和请求异常 |
......@@ -155,7 +159,9 @@ pnpm meyer-imaging -- --config=/absolute/meyer.json --state-dir=/absolute/privat
`--tick/--watch` 使用上述每轮限额；默认固定窗口单次命令仍用于明确范围的手动采集，不受每轮限额控制。修改频率后重启 watch 生效；扫描限额会在每轮重新读取配置。所有调用仍由主进程串行执行，没有并发轮次。
scan-progress 保存固定窗口、诊所、页码、页内位置和当前页原文；scan-ledger 保存已完成页的影像 ID。达到限额不推进 cursor，下一轮从断点继续。写文件先 sync 再原子替换，并同步目录；记录事实成功后才移动行断点。若数据库成功而进程在记断点前退出，下次幂等重试，不重复生成事实。
clinics/<诊所哈希>/ 下的 cursor、scan-progress、scan-ledger、failure 保存独立水位、固定窗口、分页断点和故障退避。原文、image/bundle 快照以及根目录 pending 保持原键，pending 按完整诊所归属分组调度，既有重试时间和次数不重置。达到限额只暂停当前诊所，下次从断点继续。
首次升级在根目录锁内复制旧调度进度：旧共同水位分配给原有诊所；未完成窗口中已扫完的诊所推进到窗口末，当前诊所连同分页 ledger 迁移，后续诊所保留同一未扫窗口。迁移标记提交前不会开始采集，中断可重入。旧 cursor/scan-progress 留作迁移证据，不再代表当前总进度；以 clinics/ 和 scheduler-status 为准。不要清空旧状态目录升级。写文件先 sync 再原子替换，并同步目录；记录事实成功后才移动行断点。若数据库成功而进程在记断点前退出，下次幂等重试，不重复生成事实。
续跑会重新核对第一页和当前未完成页的数量/身份指纹。检测到插入、删除或顺序变化就重扫当前诊所的固定窗口；跨页重复也阻止推进并重扫。签名 URL 变化不影响身份指纹。**美亚没有快照游标：无法承诺任意历史回填、窗口完成后更早时间的插入或不可见删除都不遗漏。** 当前机制保证自身限额和中断不跳过数据，历史任意变化仍属于尚未实现的对账范围。
......@@ -171,7 +177,7 @@ scan-progress 保存固定窗口、诊所、页码、页内位置和当前页原
### 存储分工
- **独立状态目录**：identity、数据库目标指纹、cursor、scan-progress、scan-ledger、pending、analysis-status、image、bundle、observed、scheduler-status、access 文件；本期不新增数据库采集表。状态目录须放持久磁盘、限制权限、备份。只支持单机单写者，不用于多机器共享抢任务。
- **独立状态目录**：identity、数据库目标指纹、scheduler-layout、clinic-rotation、clinics/ 各诊所进度、pending、analysis-status、image、bundle、observed、scheduler-status、access 文件；本期不新增数据库采集表。状态目录须放持久磁盘、限制权限、备份。只支持单机单写者，不用于多机器共享抢任务。
- **PAC 共用表**：标准化事件写 patient_transactions，标准影像/诊断写 patient_facts，按 Friday 宿主、租户、品牌与患者隔离；patients 只查找，不创建或移动。原始美亚依据还保留在事务 rawPayload。
- **原片**：仍在美亚/OSS，本地只留文件 ID 和业务证据，不复制文件。
......
apps/pac-service/src/integrations/meyer/clinic-state.ts 0 → 100644
import { cp } from 'node:fs/promises';
import { join } from 'node:path';
import { randomUUID } from 'node:crypto';
import { hash, sourceTime, type MeyerConfig, type MeyerScope } from './contracts';
import type { ScanProgress } from './scan';
import { MeyerStore } from './store';
interface Layout { version: 2; legacyScopes: MeyerScope[]; migrated: boolean }
export const clinicStateStore = (store: MeyerStore, scope: MeyerScope) =>
new MeyerStore(join(store.directory, 'clinics', hash(scope.clinicId)));
/** 迁移只复制调度断点；根目录原文、快照、影像键和已有重试次数保持不动。 */
export async function prepareClinicStates(store: MeyerStore, config: MeyerConfig) {
let layout = await store.read<Layout>('scheduler-layout');
if (!layout) {
layout = { version: 2, legacyScopes: config.scopes, migrated: false };
await store.write('scheduler-layout', layout);
}
if (layout.version !== 2) throw new Error('unsupported Meyer scheduler layout');
if (layout.migrated) return;
const cursor = await store.read<{ through: string }>('cursor');
const progress = await store.read<ScanProgress>('scan-progress');
if (progress && (!Number.isInteger(progress.scopeIndex) || progress.scopeIndex < 0 ||
progress.scopeIndex > layout.legacyScopes.length || !/^[a-f0-9-]{36}$/.test(progress.generation))) {
throw new Error('invalid legacy scan checkpoint; migration stopped');
}
for (const [index, scope] of layout.legacyScopes.entries()) {
const target = clinicStateStore(store, scope);
await target.write('scope', scope);
if (cursor) await target.write('cursor', cursor);
if (progress) {
if (index < progress.scopeIndex) await target.write('cursor', { through:
cursor && sourceTime(cursor.through) >= sourceTime(progress.end) ? cursor.through : progress.end });
else if (index === progress.scopeIndex) {
// ledger 内的旧 scopeIndex 改挂到单诊所的 0；缺失的必要 ledger 继续由扫描器拒绝。
try { await cp(join(store.directory, 'scan-ledger', progress.generation, String(index)),
join(target.directory, 'scan-ledger', progress.generation, '0'), { recursive: true }); }
catch (e) { if ((e as NodeJS.ErrnoException).code !== 'ENOENT') throw e; }
await target.write('scan-progress', { ...progress, scopeIndex: 0 });
} else await target.write('scan-progress', { version: 1, generation: randomUUID(),
start: progress.start, end: progress.end, scopeIndex: 0, page: 1, row: 0, pageSize: progress.pageSize });
}
}
// 标记完成后才允许任何诊所运行；迁移中断时可按冻结的 legacyScopes 幂等复制。
await store.write('scheduler-layout', { ...layout, migrated: true });
}
apps/pac-service/src/integrations/meyer/collector.ts
......@@ -115,6 +115,10 @@ export async function collectMeyer(config: MeyerConfig, source: MeyerSource, sto
await store.write(`deferred-${key}`, { reason: 'unbound patient', observedAt, image });
result.deferred++; continue;
}
if (source.canRequestAnalysis?.() === false) {
retry.reason = 'analysis_pending';
result.deferred++; continue;
}
retry.analysisAttempts!++;
retry.reason = 'analysis_pending';
retry.nextRetryAt = nextAnalysisRetry(config, image, now);
......
apps/pac-service/src/integrations/meyer/config.example.json
......@@ -17,5 +17,7 @@
"errorBackoffMaxMs": 3600000,
"clinicIds": [
"replace-clinic-id"
]
],
"maxClinicsPerTick": 5,
"maxSourceRequestsPerTick": 100
}
apps/pac-service/src/integrations/meyer/contracts.ts
......@@ -38,6 +38,8 @@ const ConfigFields = z.object({
roundTimeoutMs: z.number().int().min(1000).max(86400000).default(600000),
pollIntervalMs: z.number().int().min(1000).max(86400000).default(900000),
maxNewImagesPerTick: z.number().int().min(1).max(10000).default(100),
maxClinicsPerTick: z.number().int().min(1).max(100).default(5),
maxSourceRequestsPerTick: z.number().int().min(2).max(10000).default(100),
maxRetriesPerTick: z.number().int().min(0).max(1000).default(10),
analysisSlowAfterMs: z.number().int().min(1000).max(86400000).default(900000),
analysisSlowRetryMs: z.number().int().min(1000).max(86400000).default(3600000),
......@@ -96,6 +98,8 @@ export interface MeyerSink { apply(bundle: MeyerBundle): Promise<void> }
export interface MeyerSource {
list(scope: MeyerScope, page: number): Promise<{ total: number; page: number; size: number; images: unknown[] }>;
symptom(image: MeyerImage): Promise<unknown>;
/** 调度预算不足时延后分析，不能虚增实际分析请求次数。 */
canRequestAnalysis?(): boolean;
}
export function hash(value: unknown): string {
const stable = (v: unknown): unknown => Array.isArray(v) ? v.map(stable) :
......
apps/pac-service/src/integrations/meyer/org-mapping.ts
......@@ -44,3 +44,20 @@ export async function loadMeyerScopes(clinicIds: string[]): Promise<MeyerScope[]
throw error;
} finally { await db.$disconnect(); }
}
/** 定时采集隔离单诊所映射异常；数据库等共享依赖故障仍终止本轮。 */
export async function loadMeyerScopeResults(clinicIds: string[]) {
if (!process.env.DATABASE_URL) throw new Error('DATABASE_URL required for automatic clinic mapping');
const db = new PrismaClient();
const scopes: MeyerScope[] = [], failures: Array<{ clinicId: string; reason: string }> = [];
try {
for (const clinicId of clinicIds) {
try { scopes.push(...await resolveMeyerScopes(db, [clinicId])); }
catch (e) {
if (!(e instanceof MappingError)) throw new Error('PAC organization lookup failed');
failures.push({ clinicId, reason: 'organization mapping missing, conflicting or invalid' });
}
}
return { scopes, failures };
} finally { await db.$disconnect(); }
}
apps/pac-service/src/integrations/meyer/scan.ts
......@@ -3,6 +3,7 @@ import { join } from 'node:path';
import { collectMeyer, type CollectResult } from './collector';
import { hash, type MeyerConfig, type MeyerScope, type MeyerSource, type MeyerSink } from './contracts';
import { MeyerStore } from './store';
import { MeyerBudgetExhausted } from './tick-budget';
type Page = Awaited<ReturnType<MeyerSource['list']>>;
export interface ScanProgress {
......@@ -17,7 +18,7 @@ function checkPage(batch: Page, page: number, size: number) {
}
/** 已读取页先落盘；逐行成功/已持久化待处理后才移动页内位置。 */
export async function scanWindow(config: MeyerConfig, sourceFor: (c: MeyerConfig) => MeyerSource,
store: MeyerStore, codes: Record<string, string>, sink: MeyerSink, start: string, end: string, now = Date.now()) {
store: MeyerStore, codes: Record<string, string>, sink: MeyerSink, start: string, end: string, now = Date.now(), evidenceStore = store) {
let progress = await store.read<ScanProgress>('scan-progress');
if (!progress) {
progress = { version: 1, generation: randomUUID(), start, end, scopeIndex: 0, page: 1, row: 0, pageSize: config.pageSize };
......@@ -97,8 +98,9 @@ export async function scanWindow(config: MeyerConfig, sourceFor: (c: MeyerConfig
seen.add(ids(image));
const one = await collectMeyer({ ...window, scopes: [scope] }, {
list: async () => ({ total: 1, page: 1, size: window.pageSize, images: [image] }),
canRequestAnalysis: () => source.canRequestAnalysis?.() ?? true,
symptom: im => source.symptom(im),
}, store, codes, sink, true, now);
}, evidenceStore, codes, sink, true, now);
for (const key of ['seen', 'collected', 'applied', 'alreadyApplied', 'deferred', 'failed', 'imagesApplied', 'analysisFailed'] as const) result[key] += one[key];
result.errors.push(...one.errors);
if (one.failed) return result; // 不跳过未知失败；重试从本行恢复。
......@@ -116,7 +118,9 @@ export async function scanWindow(config: MeyerConfig, sourceFor: (c: MeyerConfig
}
result.complete = progress.scopeIndex === config.scopes.length;
} catch (e) {
if (!(e instanceof MeyerBudgetExhausted)) {
result.failed++; result.errors.push({ key: 'page-scan', reason: e instanceof Error ? e.message : 'scan failed' });
}
}
return result;
}
apps/pac-service/src/integrations/meyer/scheduler.ts
......@@ -3,14 +3,17 @@ import { rm } from 'node:fs/promises';
import { join } from 'node:path';
import { collectMeyer, type PendingImage } from './collector';
import { MeyerStore, type SavedBundle } from './store';
import { imageKey, sourceTime, type MeyerConfig, type MeyerSource, type MeyerSink } from './contracts';
import { imageKey, sourceTime, type MeyerConfig, type MeyerSource, type MeyerSink, hash, type MeyerScope } from './contracts';
import { analysisCapability } from './retry-policy';
import { clinicStateStore, prepareClinicStates } from './clinic-state';
import { MeyerBudgetExhausted, MeyerTickBudget } from './tick-budget';
export const chinaTime = (ms: number) => new Date(ms + 8 * 3600000).toISOString().slice(0, 19).replace('T', ' ');
/** 一轮独立任务：先限量重试，再扫描一个窗口；只推进本适配器水位。 */
export async function runMeyerTick(config: MeyerConfig, sourceFor: (c: MeyerConfig) => MeyerSource,
store: MeyerStore, codes: Record<string, string>, sink: MeyerSink, now = Date.now()) {
const cursor = await store.read<{ through: string }>('cursor');
async function runClinicTick(config: MeyerConfig, sourceFor: (c: MeyerConfig) => MeyerSource,
store: MeyerStore, progressStore: MeyerStore, pendingKeys: string[], codes: Record<string, string>, sink: MeyerSink,
available: () => boolean, now: number) {
const cursor = await progressStore.read<{ through: string }>('cursor');
const initial = Date.parse(sourceTime(config.timeStart));
const through = cursor ? Date.parse(sourceTime(cursor.through)) : initial;
if (!Number.isFinite(through) || through < initial) throw new Error('invalid collector cursor');
......@@ -19,10 +22,10 @@ export async function runMeyerTick(config: MeyerConfig, sourceFor: (c: MeyerConf
const pending = new MeyerStore(join(store.directory, 'pending'));
const due: Array<{ key: string; value: PendingImage }> = [];
let retiredUnsupported = 0;
for (const key of await pending.keys()) {
for (const key of pendingKeys) {
const value = await pending.read<PendingImage>(key);
if (!value) continue;
if (!config.scopes.some(s => JSON.stringify(s) === JSON.stringify(value.scope))) throw new Error('pending scope mismatch');
if (hash(value.scope) !== hash(config.scopes[0])) throw new Error('pending scope mismatch');
const saved = await store.read<SavedBundle>(`bundle-${key}`);
// 旧队列只清理已成功落库、且没有待写分析快照的非分析类型。
// 不再访问美亚，也不消耗每轮重试名额；归属和命名空间仍由 collector 校验。
......@@ -41,9 +44,11 @@ export async function runMeyerTick(config: MeyerConfig, sourceFor: (c: MeyerConf
if ((value.autoRetry !== false || saved) && Date.parse(value.nextRetryAt) <= now) due.push({ key, value });
}
due.sort((a, b) => a.value.nextRetryAt.localeCompare(b.value.nextRetryAt));
let retryFailed = 0;
let retryFailed = 0, retryAttempted = 0;
for (const { key, value } of due.slice(0, config.maxRetriesPerTick)) {
if (!config.scopes.some(s => JSON.stringify(s) === JSON.stringify(value.scope))) throw new Error('pending scope mismatch');
if (!available()) break;
retryAttempted++;
if (hash(value.scope) !== hash(config.scopes[0])) throw new Error('pending scope mismatch');
if (!value.image.imageCreateTime) { retryFailed++; continue; }
const t = Date.parse(sourceTime(value.image.imageCreateTime));
const retryConfig = { ...config, scopes: [value.scope], timeStart: chinaTime(t - 1000), timeEnd: chinaTime(t + 1000) };
......@@ -69,10 +74,12 @@ export async function runMeyerTick(config: MeyerConfig, sourceFor: (c: MeyerConf
}
const result = await collectMeyer(retryConfig, {
list: async () => ({ total: 1, page: 1, size: config.pageSize, images: [found] }),
canRequestAnalysis: () => client.canRequestAnalysis?.() ?? true,
symptom: image => client.symptom(image),
}, store, codes, sink, true, now);
retryFailed += result.failed + result.analysisFailed;
} catch {
} catch (error) {
if (error instanceof MeyerBudgetExhausted) break;
retryFailed++;
// collector 可能已持久化请求次数后才在写运行摘要时失败；不能回写旧次数。
const retained = await pending.read<PendingImage>(key) ?? value;
......@@ -81,29 +88,146 @@ export async function runMeyerTick(config: MeyerConfig, sourceFor: (c: MeyerConf
}
}
let scan: Awaited<ReturnType<typeof scanWindow>> | undefined;
let progress = await store.read<ScanProgress>('scan-progress');
let progress = await progressStore.read<ScanProgress>('scan-progress');
// 上轮水位已提交但进程尚未删 checkpoint 就退出，可安全收尾。
if (progress && Date.parse(sourceTime(progress.end)) <= through) {
await store.remove('scan-progress');
await rm(join(store.directory, 'scan-ledger'), { recursive: true, force: true });
await progressStore.remove('scan-progress');
await rm(join(progressStore.directory, 'scan-ledger'), { recursive: true, force: true });
progress = null;
}
if (progress || end > through) {
const start = progress?.start ?? chinaTime(Math.max(initial, through - 60000));
const finish = progress?.end ?? chinaTime(end);
scan = await scanWindow(config, sourceFor, store, codes, sink, start, finish, now);
scan = await scanWindow(config, sourceFor, progressStore, codes, sink, start, finish, now, store);
if (scan.complete && !scan.failed) {
await store.write('cursor', { through: finish });
await store.remove('scan-progress');
await rm(join(store.directory, 'scan-ledger'), { recursive: true, force: true });
await progressStore.write('cursor', { through: finish });
await progressStore.remove('scan-progress');
await rm(join(progressStore.directory, 'scan-ledger'), { recursive: true, force: true });
}
}
let manualReview = 0;
const keys = await pending.keys();
const keys = [];
for (const key of pendingKeys) if (await pending.read(key)) keys.push(key);
// 新扫描产生的队列项在整轮汇总时再统一统计。
for (const key of keys) if ((await pending.read<PendingImage>(key))?.autoRetry === false) manualReview++;
const status = { checkedAt: new Date(now).toISOString(), retryAttempted: Math.min(config.maxRetriesPerTick, due.length), retryFailed,
const status = { checkedAt: new Date(now).toISOString(), retryAttempted, retryFailed,
pending: keys.length, manualReview, autoPending: keys.length - manualReview, retiredUnsupported,
scan, cursor: await store.read('cursor') };
scan, cursor: await progressStore.read<{ through: string }>('cursor') };
await progressStore.write('scheduler-status', status);
return status;
}
interface ClinicRun {
clinicId: string; failed: boolean; skipped?: 'backoff' | 'budget'; nextRunAt?: string;
cursor?: { through: string } | null; requests: number; pending?: number; manualReview?: number;
}
/** 单一定时器、串行轮转；进度和故障退避归诊所，业务请求与重试额度归整轮。 */
export async function runMeyerTick(config: MeyerConfig, sourceFor: (c: MeyerConfig) => MeyerSource,
store: MeyerStore, codes: Record<string, string>, sink: MeyerSink, now = Date.now()) {
await prepareClinicStates(store, config);
const scopes = [...config.scopes].sort((a, b) => a.clinicId.localeCompare(b.clinicId));
const rotation = await store.read<{ nextClinicId: string }>('clinic-rotation');
const start = Math.max(0, scopes.findIndex(s => s.clinicId === rotation?.nextClinicId));
const ordered = [...scopes.slice(start), ...scopes.slice(0, start)];
const pending = new MeyerStore(join(store.directory, 'pending'));
const grouped = new Map<string, string[]>();
let queueErrors = 0;
for (const key of await pending.keys()) {
try {
const value = await pending.read<PendingImage>(key);
if (!value) continue;
const scopeKey = hash(value.scope);
const group = grouped.get(scopeKey) ?? [];
group.push(key); grouped.set(scopeKey, group);
} catch { queueErrors++; } // 单个损坏条目留底，其他诊所仍可运行。
}
const budget = new MeyerTickBudget(config.maxSourceRequestsPerTick,
Math.max(1, config.roundTimeoutMs - config.timeoutMs - 1000));
let retriesLeft = config.maxRetriesPerTick, imagesLeft = config.maxNewImagesPerTick,
pagesLeft = config.maxListPagesPerTick;
const visitLimit = Math.min(scopes.length, config.maxClinicsPerTick, config.maxNewImagesPerTick,
Math.max(1, Math.floor(config.maxListPagesPerTick / 2)), Math.max(1, Math.floor(budget.limit / 2)));
const clinics: ClinicRun[] = [];
const aggregate = { pages: 0, seen: 0, collected: 0, applied: 0, alreadyApplied: 0, deferred: 0,
failed: 0, imagesApplied: 0, analysisFailed: 0, examined: 0, complete: true };
let retryAttempted = 0, retryFailed = 0, retiredUnsupported = 0, clinicFailures = 0;
for (const [index, scope] of ordered.slice(0, visitLimit).entries()) {
if (!budget.available()) break;
const state = clinicStateStore(store, scope);
// 在请求前推进轮转指针：进程中断也不能让同一家永久占据队首。
await store.write('clinic-rotation', { nextClinicId: ordered[(index + 1) % ordered.length]!.clinicId });
const beforeRequests = budget.requests;
try {
const previousScope = await state.read<MeyerScope>('scope');
if (previousScope && hash(previousScope) !== hash(scope)) throw new Error('clinic ownership mapping changed');
if (!previousScope) await state.write('scope', scope);
const failure = await state.read<{ consecutiveFailures: number; nextRunAt: string }>('failure');
if (failure && Date.parse(failure.nextRunAt) > now) {
clinics.push({ clinicId: scope.clinicId, failed: false, skipped: 'backoff', requests: 0,
nextRunAt: failure.nextRunAt, cursor: await state.read('cursor') });
aggregate.complete = false; continue;
}
const remaining = visitLimit - index;
const allocation = budget.source(sourceFor, Math.max(2, Math.floor((budget.limit - budget.requests) / remaining)));
const quota = { ...config, scopes: [scope],
maxRetriesPerTick: Math.ceil(retriesLeft / remaining),
maxNewImagesPerTick: Math.ceil(imagesLeft / remaining),
maxListPagesPerTick: Math.floor(pagesLeft / remaining) };
const result = await runClinicTick(quota, allocation.factory, store, state,
grouped.get(hash(scope)) ?? [], codes, sink, allocation.available, now);
retriesLeft -= result.retryAttempted;
retryAttempted += result.retryAttempted; retryFailed += result.retryFailed;
retiredUnsupported += result.retiredUnsupported;
if (result.scan) {
pagesLeft -= result.scan.pages; imagesLeft -= result.scan.examined;
for (const key of ['pages', 'seen', 'collected', 'applied', 'alreadyApplied', 'deferred', 'failed',
'imagesApplied', 'analysisFailed', 'examined'] as const) aggregate[key] += result.scan[key];
}
aggregate.complete &&= result.scan?.complete ?? true;
const failed = !!(result.retryFailed || result.scan?.failed || result.scan?.analysisFailed);
if (result.scan?.failed) {
const consecutiveFailures = (failure?.consecutiveFailures ?? 0) + 1;
const delay = Math.min(config.errorBackoffMaxMs, config.pollIntervalMs * 2 ** Math.min(10, consecutiveFailures - 1));
await state.write('failure', { consecutiveFailures, nextRunAt: new Date(now + delay).toISOString() });
} else await state.remove('failure');
// 分析/单条补查失败使用该条目的 nextRetryAt，不降低新影像扫描频率。
if (failed) clinicFailures++;
clinics.push({ clinicId: scope.clinicId, failed, requests: budget.requests - beforeRequests, cursor: result.cursor });
} catch {
// 无法读取该诊所状态也不猜水位；保留文件，继续其他诊所。
clinicFailures++; aggregate.complete = false;
clinics.push({ clinicId: scope.clinicId, failed: true, requests: budget.requests - beforeRequests });
}
}
let total = 0, manualReview = 0;
const queueCounts = new Map<string, { pending: number; manualReview: number }>();
const activeScopes = new Set(scopes.map(hash));
for (const key of await pending.keys()) {
try {
const value = await pending.read<PendingImage>(key);
if (!value || !activeScopes.has(hash(value.scope))) continue;
total++; if (value.autoRetry === false) manualReview++;
const counts = queueCounts.get(value.scope.clinicId) ?? { pending: 0, manualReview: 0 };
counts.pending++; if (value.autoRetry === false) counts.manualReview++;
queueCounts.set(value.scope.clinicId, counts);
} catch { /* 已在 queueErrors 报告；不覆盖损坏原文。 */ }
}
for (const clinic of clinics) {
Object.assign(clinic, queueCounts.get(clinic.clinicId) ?? { pending: 0, manualReview: 0 });
if (clinic.failed || clinic.skipped) continue;
const scope = scopes.find(s => s.clinicId === clinic.clinicId)!;
const state = clinicStateStore(store, scope);
try {
const previous = await state.read<Record<string, unknown>>('scheduler-status');
if (previous) await state.write('scheduler-status', { ...previous, pending: clinic.pending,
manualReview: clinic.manualReview, autoPending: clinic.pending! - clinic.manualReview! });
} catch { clinic.failed = true; clinicFailures++; }
}
const status = { checkedAt: new Date(now).toISOString(), pending: total, manualReview, autoPending: total - manualReview,
retryAttempted, retryFailed, retiredUnsupported, clinicFailures, queueErrors, sourceRequests: budget.requests,
clinics, scan: { ...aggregate, complete: aggregate.complete && clinics.length === scopes.length },
cursor: scopes.length === 1 ? clinics[0]?.cursor ?? await clinicStateStore(store, scopes[0]!).read<{ through: string }>('cursor') : null };
await store.write('scheduler-status', status);
return status;
}
apps/pac-service/src/integrations/meyer/store.ts
......@@ -7,7 +7,7 @@ import { hash, type MeyerBundle } from './contracts';
/** 独立磁盘留底，不写主摄入游标。目录只能被单个 CLI 占用，文件默认仅当前用户可读。 */
export class MeyerStore {
constructor(readonly directory: string) {}
async acquire(identity: unknown): Promise<() => Promise<void>> {
async acquire(identity: unknown, legacyIdentities: unknown[] = [], migrate?: () => Promise<void>): Promise<() => Promise<void>> {
await mkdir(this.directory, { recursive: true, mode: 0o700 });
const lock = join(this.directory, '.lock');
try { await mkdir(lock, { mode: 0o700 }); } catch {
......@@ -33,8 +33,13 @@ export class MeyerStore {
await writeFile(join(lock, 'pid'), String(process.pid), { mode: 0o600 });
const expected = hash(identity);
const existing = await this.read<{ identity: string }>('identity');
if (existing && existing.identity !== expected) throw new Error('Meyer state scope/config mismatch; use its original configuration');
if (!existing) await this.write('identity', { identity: expected });
if (existing && existing.identity !== expected && !legacyIdentities.some(v => hash(v) === existing.identity)) {
throw new Error('Meyer state scope/config mismatch; use its original configuration');
}
if (!existing || existing.identity !== expected) {
await migrate?.();
await this.write('identity', { identity: expected });
}
return () => rm(lock, { recursive: true });
} catch (error) { await rm(lock, { recursive: true }); throw error; }
}
......
apps/pac-service/src/integrations/meyer/tick-budget.ts 0 → 100644
import type { MeyerConfig, MeyerSource } from './contracts';
export class MeyerBudgetExhausted extends Error {}
/** 整轮与诊所配额同时扣减；只统计列表/分析业务请求，不含令牌申请。 */
export class MeyerTickBudget {
requests = 0;
readonly deadline: number;
constructor(readonly limit: number, durationMs: number) { this.deadline = Date.now() + durationMs; }
available() { return this.requests < this.limit && Date.now() < this.deadline; }
source(factory: (config: MeyerConfig) => MeyerSource, allowance: number) {
let used = 0;
const available = () => used < allowance && this.available();
const spend = () => {
if (!available()) throw new MeyerBudgetExhausted('Meyer tick budget exhausted');
used++; this.requests++;
};
return { available, factory: (config: MeyerConfig): MeyerSource => {
const source = factory(config);
return {
list: async (scope, page) => { spend(); return source.list(scope, page); },
canRequestAnalysis: available,
symptom: async image => { spend(); return source.symptom(image); },
};
} };
}
}
apps/pac-service/tests/meyer-analysis-retry.spec.ts
......@@ -5,6 +5,7 @@ import { collectMeyer, type PendingImage } from '../src/integrations/meyer/colle
import { imageKey, type MeyerSource } from '../src/integrations/meyer/contracts';
import { analysisCapability } from '../src/integrations/meyer/retry-policy';
import { runMeyerTick, chinaTime } from '../src/integrations/meyer/scheduler';
import { clinicStateStore } from '../src/integrations/meyer/clinic-state';
import { MeyerStore } from '../src/integrations/meyer/store';
import { meyerConfig, meyerImage, meyerSymptom, codes } from './meyer-imaging-fixtures';
......@@ -130,7 +131,7 @@ describe('Meyer analysis retry policy', () => {
expect(result.retryFailed).toBe(1);
expect(await s.pending.read(s.key)).toMatchObject({ analysisAttempts: 1, autoRetry: false, reason: 'analysis_exhausted' });
expect(s.source.symptom).toHaveBeenCalledTimes(1);
await s.store.write('cursor', { through: chinaTime(created + 4 * 3600000) });
await clinicStateStore(s.store, s.config.scopes[0]!).write('cursor', { through: chinaTime(created + 4 * 3600000) });
const factory = jest.fn(() => s.source);
await runMeyerTick(s.config, factory, s.store, codes, s.sink, created + 4 * 3600000);
expect(factory).not.toHaveBeenCalled();
......
apps/pac-service/tests/meyer-clinic-isolation.spec.ts 0 → 100644
import { mkdtemp, rm } from 'node:fs/promises';
import { join } from 'node:path';
import { tmpdir } from 'node:os';
import { randomUUID } from 'node:crypto';
import { ConfigSchema, hash, imageKey, type MeyerConfig, type MeyerScope, type MeyerSource } from '../src/integrations/meyer/contracts';
import { MeyerStore } from '../src/integrations/meyer/store';
import { clinicStateStore, prepareClinicStates } from '../src/integrations/meyer/clinic-state';
import { runMeyerTick } from '../src/integrations/meyer/scheduler';
import { meyerConfig, meyerImage, meyerSymptom, codes } from './meyer-imaging-fixtures';
describe('Meyer clinic isolation and shared quotas', () => {
let dir: string, store: MeyerStore;
const now = Date.parse('2026-10-10T03:00:00Z');
const scopes = (count: number): MeyerScope[] => Array.from({ length: count }, (_, i) => ({
clinicId: `clinic-${String(i).padStart(3, '0')}`, tenantId: 'friday-market', sourceUnit: 'brand-a' }));
const config = (count = 2) => ConfigSchema.parse({ ...meyerConfig(), scopes: scopes(count), timeStart: '2026-10-10 09:00:00' });
const empty: MeyerSource = { list: async (_, page) => ({ total: 0, page, size: 100, images: [] }), symptom: async () => meyerSymptom() };
beforeEach(async () => { dir = await mkdtemp(join(tmpdir(), 'meyer-clinic-')); store = new MeyerStore(dir); });
afterEach(async () => { await rm(dir, { recursive: true, force: true }); });
const sink = { apply: async () => {} };
const state = (c: MeyerConfig, index = 0) => clinicStateStore(store, c.scopes[index]!);
test('failing clinic keeps its cursor and backs off; another clinic advances in the same tick', async () => {
const c = config(), seen: string[] = [];
const factory = (): MeyerSource => ({ ...empty, list: async (scope, page) => {
seen.push(scope.clinicId); if (scope.clinicId === c.scopes[0]!.clinicId) throw Error('clinic unavailable');
return empty.list(scope, page);
} });
const result = await runMeyerTick(c, factory, store, codes, sink, now);
expect(result.clinicFailures).toBe(1);
expect(await state(c).read('cursor')).toBeNull();
expect(await state(c, 1).read('cursor')).toEqual({ through: '2026-10-10 10:00:00' });
expect(seen).toEqual(c.scopes.map(s => s.clinicId));
seen.length = 0;
const later = await runMeyerTick(c, factory, store, codes, sink, now + 1000);
expect(later.clinics.find(x => x.clinicId === c.scopes[0]!.clinicId)?.skipped).toBe('backoff');
expect(seen).toEqual([c.scopes[1]!.clinicId]);
});
test('analysis outage uses item retry times and does not back off new image scans', async () => {
const c = config(1); let pages = 0;
const factory = (window: MeyerConfig): MeyerSource => ({
list: async scope => { pages++; const images = window.timeStart <= meyerImage().imageCreateTime!
? [{ ...meyerImage(), clinicId: scope.clinicId }] : [];
return { total: images.length, page: 1, size: 100, images }; },
symptom: async () => { throw Error('analysis unavailable'); },
});
const first = await runMeyerTick(c, factory, store, codes, sink, now);
expect(first.scan.analysisFailed).toBe(1);
expect(await state(c).read('failure')).toBeNull();
const second = await runMeyerTick(c, factory, store, codes, sink, now + c.pollIntervalMs);
expect(second.clinics[0]?.skipped).toBeUndefined(); expect(pages).toBe(2);
});
test('100 clinics share ten retries per round and rotate without starving later clinics', async () => {
const c = config(100), pending = new MeyerStore(join(dir, 'pending'));
for (const scope of c.scopes) {
const im = { ...meyerImage(), clinicId: scope.clinicId };
await pending.write(imageKey(scope, im.imageId), { scope, image: im, attempts: 1,
analysisAttempts: 2, nextRetryAt: '2000-01-01T00:00:00Z' });
}
const seen = new Set<string>();
for (let i = 0; i < 20; i++) {
const result = await runMeyerTick(c, () => empty, store, codes, sink, now + i * c.pollIntervalMs);
expect(result.retryAttempted).toBeLessThanOrEqual(10);
expect(result.sourceRequests).toBeLessThanOrEqual(100);
expect(result.scan.pages).toBeLessThanOrEqual(10);
expect(result.clinics.length).toBeLessThanOrEqual(5);
for (const x of result.clinics) seen.add(x.clinicId);
}
expect(seen.size).toBe(100);
expect((await pending.keys()).length).toBe(100);
}, 60000);
test('tiny image budget still rotates across clinics; long first-clinic history cannot monopolize scans', async () => {
const c = { ...config(), maxNewImagesPerTick: 1 }, written: string[] = [];
const factory = (): MeyerSource => ({ list: async scope => ({ total: 3, page: 1, size: 100,
images: [0, 1, 2].map(n => ({ ...meyerImage(), imageId: `im-${n}`, clinicId: scope.clinicId, aiType: 32 })) }),
symptom: async () => { throw Error('unsupported'); } });
for (let i = 0; i < 2; i++) await runMeyerTick(c, factory, store, codes,
{ apply: async b => { written.push(b.scope.clinicId); } }, now);
expect(written).toEqual(c.scopes.map(s => s.clinicId));
for (let i = 0; i < 2; i++) expect(await state(c, i).read('scan-progress')).toMatchObject({ row: 1 });
});
test('request quota defers analysis without consuming an actual analysis attempt or dropping metadata', async () => {
const c = { ...config(1), maxSourceRequestsPerTick: 2 }, written: string[] = [];
const images = [1, 2].map(n => ({ ...meyerImage(), imageId: `im-${n}`, clinicId: c.scopes[0]!.clinicId }));
const analysis = jest.fn(async () => ({ teeth: [], diagnosis: '', description: '' }));
const result = await runMeyerTick(c, () => ({ list: async () => ({ total: 2, page: 1, size: 100, images }),
symptom: analysis }), store, codes, { apply: async b => { written.push(b.raw.image.imageId); } }, now);
expect(result.sourceRequests).toBe(2); expect(result.clinicFailures).toBe(0);
expect(written).toHaveLength(2); expect(analysis).toHaveBeenCalledTimes(1);
const pending = new MeyerStore(join(dir, 'pending'));
expect(await pending.read(imageKey(c.scopes[0]!, images[1]!.imageId))).toMatchObject({ analysisAttempts: 0, autoRetry: true });
});
test('migration retains earlier completed clinics, current page ledger and later unscanned window', async () => {
const c = config(3), generation = randomUUID();
await store.write('cursor', { through: '2026-10-10 09:00:00' });
await store.write('scan-progress', { version: 1, generation, start: '2026-10-10 09:00:00', end: '2026-10-10 10:00:00',
scopeIndex: 1, page: 2, row: 1, pageSize: 100 });
const ledger = new MeyerStore(join(dir, 'scan-ledger', generation, '1'));
await ledger.write('page-1', ['legacy-image']);
await prepareClinicStates(store, c);
expect(await state(c).read('cursor')).toEqual({ through: '2026-10-10 10:00:00' });
expect(await state(c, 1).read('scan-progress')).toMatchObject({ scopeIndex: 0, page: 2, row: 1, generation });
expect(await new MeyerStore(join(state(c, 1).directory, 'scan-ledger', generation, '0')).read('page-1')).toEqual(['legacy-image']);
expect(await state(c, 2).read('scan-progress')).toMatchObject({ page: 1, row: 0, end: '2026-10-10 10:00:00' });
await state(c).write('cursor', { through: '2026-10-11 00:00:00' });
await prepareClinicStates(store, c);
expect(await state(c).read('cursor')).toEqual({ through: '2026-10-11 00:00:00' });
});
test('added clinics start at configured history, not another clinic cursor; removed queues are retained', async () => {
const c = config(1); await store.write('cursor', { through: '2026-10-10 10:00:00' });
await prepareClinicStates(store, c);
const extra = config(2), windows: Record<string, string> = {};
await runMeyerTick(extra, window => ({ ...empty, list: async (s, page) => {
windows[s.clinicId] = window.timeStart; return empty.list(s, page);
} }), store, codes, sink, now);
expect(windows[c.scopes[0]!.clinicId]).toBe('2026-10-10 09:59:00');
expect(windows[extra.scopes[1]!.clinicId]).toBe('2026-10-10 09:00:00');
const pending = new MeyerStore(join(dir, 'pending'));
await pending.write('retired-clinic', { scope: extra.scopes[1], image: meyerImage(), attempts: 2, nextRetryAt: '2000-01-01T00:00:00Z' });
const result = await runMeyerTick(c, () => empty, store, codes, sink, now);
expect(result.retryAttempted).toBe(0); expect(await pending.read('retired-clinic')).not.toBeNull();
});
test('ownership changes and corrupt state fail only the affected clinic', async () => {
const c = config(); await prepareClinicStates(store, c);
await state(c).write('scope', { ...c.scopes[0], sourceUnit: 'another-brand' });
const result = await runMeyerTick(c, () => empty, store, codes, sink, now);
expect(result.clinicFailures).toBe(1);
expect(await state(c).read('cursor')).toBeNull();
expect(await state(c, 1).read('cursor')).not.toBeNull();
});
test('identity upgrade is locked and accepts only an exact legacy identity', async () => {
const old = { account: 'a', scopes: config().scopes }, next = { version: 2, account: 'a' };
const release = await store.acquire(old); await release();
const migrated = jest.fn(async () => { expect(await store.read('identity')).toEqual({ identity: hash(old) }); });
await expect(store.acquire(next, [{ ...old, account: 'b' }], migrated)).rejects.toThrow('scope/config');
expect(migrated).not.toHaveBeenCalled();
const done = await store.acquire(next, [old], migrated); await done();
expect(migrated).toHaveBeenCalledTimes(1);
const again = await store.acquire(next, [], migrated); await again();
expect(migrated).toHaveBeenCalledTimes(1);
});
});
apps/pac-service/tests/meyer-imaging-postgres.spec.ts
import { createMeyerOpenHttp } from './helpers/meyer-open-http';
import { hashSecret } from '@pac/utils';
import { PatientFactsLookupResponseSchema } from '../src/modules/open/patient-facts-lookup.schema';
import { resolveMeyerScopes } from '../src/integrations/meyer/org-mapping';
import { resolveMeyerScopes, loadMeyerScopeResults } from '../src/integrations/meyer/org-mapping';
import { InputConfigSchema } from '../src/integrations/meyer/contracts';
import { NestFactory } from '@nestjs/core';
import type { INestApplicationContext } from '@nestjs/common';
......@@ -68,6 +68,19 @@ describePg('Meyer | isolated PostgreSQL + actual minimal CLI module', () => {
clinicId: 'clinic-a', action: 'visit_registered', subjectType: 'visit', subjectId: 'host-visit',
rawPayload: {}, payloadHash: 'test', occurredAt: new Date() } });
}
test('automatic clinic mapping isolates missing and conflicting clinics without losing valid scopes', async () => {
await hostRecord();
const result = await loadMeyerScopeResults(['missing-clinic', 'clinic-a']);
expect(result.scopes).toEqual(meyerConfig().scopes);
expect(result.failures).toEqual([{ clinicId: 'missing-clinic', reason: expect.any(String) }]);
await prisma.patientTransaction.create({ data: { hostId, tenantId: 'friday-market', patientId: otherPatientId,
clinicId: 'conflicting-clinic', action: 'visit_registered', subjectType: 'visit', subjectId: 'other', rawPayload: {}, payloadHash: 'test', occurredAt: new Date() } });
await prisma.patientTransaction.create({ data: { hostId, tenantId: 'friday-market', patientId,
clinicId: 'conflicting-clinic', action: 'visit_registered', subjectType: 'visit', subjectId: 'another', rawPayload: {}, payloadHash: 'test', occurredAt: new Date() } });
const next = await loadMeyerScopeResults(['conflicting-clinic', 'clinic-a']);
expect(next.scopes).toEqual(meyerConfig().scopes);
expect(next.failures.map(x => x.clinicId)).toEqual(['conflicting-clinic']);
});
test('file refs backfill preserves versions, watermarks, frozen evidence and analysis; subsequent replay is unchanged', async () => {
const bundle = meyerBundle();
await sink.apply(bundle);
......
apps/pac-service/tests/meyer-resume.spec.ts
......@@ -2,6 +2,7 @@ import { mkdtemp, rm, mkdir, writeFile } from 'node:fs/promises';
import { hostname, tmpdir } from 'node:os';
import { spawn } from 'node:child_process';
import { join } from 'node:path';
import { clinicStateStore } from '../src/integrations/meyer/clinic-state';
import { MeyerStore } from '../src/integrations/meyer/store';
import { runMeyerTick } from '../src/integrations/meyer/scheduler';
import { watchMeyer, withRoundDeadline } from '../src/integrations/meyer/watch';
......@@ -24,7 +25,7 @@ describe('Meyer conservative limits and recovery', () => {
const sink = { apply: async (b: MeyerBundle) => { if (b.phase === 'analysis') applied.push(b.raw.image.imageId); } };
const tick = () => runMeyerTick(c, () => source(rows), new MeyerStore(dir), codes, sink, now);
const first = await tick(); expect(first.scan?.examined).toBe(2); expect(first.cursor).toBeNull();
expect(await new MeyerStore(dir).read('scan-progress')).toMatchObject({ page: 1, row: 2 });
expect(await clinicStateStore(new MeyerStore(dir), c.scopes[0]!).read('scan-progress')).toMatchObject({ page: 1, row: 2 });
rows.forEach(r => { Object.assign(r, { imageUrl: 'https://example.com/renewed-signature' }); });
const second = await tick(); expect(second.scan?.examined).toBe(2); expect(second.cursor).toBeNull();
const third = await tick(); expect(third.scan?.examined).toBe(1); expect(third.scan?.complete).toBe(true);
......@@ -52,7 +53,7 @@ describe('Meyer conservative limits and recovery', () => {
const first = await runMeyerTick(c, factory, new MeyerStore(dir), codes, sink, now);
expect(first.scan?.failed).toBe(1); expect(first.cursor).toBeNull();
down = false;
const next = await runMeyerTick(c, factory, new MeyerStore(dir), codes, sink, now);
const next = await runMeyerTick(c, factory, new MeyerStore(dir), codes, sink, now + c.pollIntervalMs);
expect(next.scan?.complete).toBe(true); expect(sink.apply).toHaveBeenCalledTimes(10);
});
test('duplicate caused by offset drift blocks completion and can recover by re-scan', async () => {
......@@ -65,7 +66,7 @@ describe('Meyer conservative limits and recovery', () => {
const first = await runMeyerTick(c, factory, new MeyerStore(dir), codes, sink, now);
expect(first.scan?.failed).toBe(1); expect(first.cursor).toBeNull();
drift = false;
const next = await runMeyerTick(c, factory, new MeyerStore(dir), codes, sink, now);
const next = await runMeyerTick(c, factory, new MeyerStore(dir), codes, sink, now + c.pollIntervalMs);
expect(next.scan?.complete).toBe(true); expect(sink.apply).toHaveBeenCalledTimes(10);
});
test('watch backs off for thrown outages and error exits, recovers, then stops cleanly', async () => {
......
apps/pac-service/tests/meyer-scheduler.spec.ts
......@@ -2,6 +2,7 @@ import { mkdtemp, rm } from 'node:fs/promises';
import { join } from 'node:path';
import { tmpdir } from 'node:os';
import { runMeyerTick, chinaTime } from '../src/integrations/meyer/scheduler';
import { clinicStateStore } from '../src/integrations/meyer/clinic-state';
import { MeyerStore } from '../src/integrations/meyer/store';
import { imageKey, type MeyerSource } from '../src/integrations/meyer/contracts';
import { parseMeyerArgs } from '../src/cli/meyer-imaging.cli';
......@@ -25,7 +26,7 @@ describe('independent Meyer scheduling', () => {
const first = await runMeyerTick(c, factory, store, codes, sink, now);
expect(first.pending).toBe(1); expect(first.scan?.failed).toBe(0); expect(sink.apply).toHaveBeenCalledTimes(1);
expect(first.scan?.imagesApplied).toBe(1);
expect(await store.read('cursor')).toEqual({ through: '2026-10-10 10:00:00' });
expect(await clinicStateStore(store, c.scopes[0]!).read('cursor')).toEqual({ through: '2026-10-10 10:00:00' });
const key = imageKey(c.scopes[0]!, im.imageId);
const item = await pending.read<any>(key); await pending.write(key, { ...item, nextRetryAt: '2000-01-01T00:00:00Z' });
ready = true;
......
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