Skip to content
Projects
Groups
Snippets
Help
This project
Loading...
Sign in / Register
Toggle navigation
P
pac
Overview
Overview
Details
Activity
Cycle Analytics
Repository
Repository
Files
Commits
Branches
Tags
Contributors
Graph
Compare
Charts
Issues
0
Issues
0
List
Board
Labels
Milestones
Merge Requests
0
Merge Requests
0
CI / CD
CI / CD
Pipelines
Jobs
Schedules
Charts
Wiki
Wiki
Snippets
Snippets
Members
Collapse sidebar
Close sidebar
Activity
Graph
Charts
Create a new issue
Jobs
Commits
Issue Boards
Open sidebar
ai-tools
pac
Commits
ff1c0399
Commit
ff1c0399
authored
Oct 21, 2026
by
luoqi
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
fix(meyer): bound identity retries and retain recoverable review records
parent
4c15d765
Pipeline
#3847
failed in 0 seconds
Changes
9
Pipelines
1
Show whitespace changes
Inline
Side-by-side
Showing
9 changed files
with
436 additions
and
21 deletions
+436
-21
apps/pac-service/src/cli/meyer-imaging.cli.ts
+23
-3
apps/pac-service/src/integrations/meyer/README.md
+15
-2
apps/pac-service/src/integrations/meyer/collector.ts
+16
-2
apps/pac-service/src/integrations/meyer/identity-retry.ts
+90
-0
apps/pac-service/src/integrations/meyer/pac-sink.ts
+7
-4
apps/pac-service/src/integrations/meyer/scheduler.ts
+29
-10
apps/pac-service/src/integrations/meyer/store.ts
+9
-0
apps/pac-service/tests/meyer-identity-retry.spec.ts
+195
-0
apps/pac-service/tests/meyer-imaging-postgres.spec.ts
+52
-0
No files found.
apps/pac-service/src/cli/meyer-imaging.cli.ts
View file @
ff1c0399
...
...
@@ -14,13 +14,18 @@ import { PrismaClient } from '@prisma/client';
import
{
loadDiagnosisCodes
}
from
'../integrations/meyer/mapper'
;
import
{
MeyerStore
,
type
SavedBundle
}
from
'../integrations/meyer/store'
;
import
type
{
MeyerSink
}
from
'../integrations/meyer/contracts'
;
import
{
resumeIdentity
}
from
'../integrations/meyer/identity-retry'
;
export
function
parseMeyerArgs
(
argv
:
string
[])
{
const
args
=
{
config
:
''
,
state
:
''
,
apply
:
false
,
help
:
false
,
tick
:
false
,
watch
:
false
,
imageFiles
:
''
,
backfillFileRefs
:
false
};
const
args
=
{
config
:
''
,
state
:
''
,
apply
:
false
,
help
:
false
,
tick
:
false
,
watch
:
false
,
imageFiles
:
''
,
backfillFileRefs
:
false
,
resumeIdentity
:
''
};
for
(
const
a
of
argv
)
{
if
(
a
===
'--'
)
continue
;
if
(
a
===
'--help'
||
a
===
'-h'
)
args
.
help
=
true
;
else
if
(
a
.
startsWith
(
'--image-files='
))
args
.
imageFiles
=
a
.
slice
(
14
);
else
if
(
a
.
startsWith
(
'--resume-identity='
))
{
args
.
resumeIdentity
=
a
.
slice
(
18
);
if
(
!
args
.
resumeIdentity
.
trim
())
throw
new
Error
(
'--resume-identity requires an image ID'
);
}
else
if
(
a
===
'--backfill-file-refs'
)
args
.
backfillFileRefs
=
true
;
else
if
(
a
===
'--tick'
)
args
.
tick
=
true
;
else
if
(
a
===
'--watch'
)
{
args
.
watch
=
true
;
args
.
tick
=
true
;
}
...
...
@@ -32,13 +37,14 @@ export function parseMeyerArgs(argv: string[]) {
if
(
!
args
.
help
&&
(
!
args
.
config
||
!
args
.
state
))
throw
new
Error
(
'--config and --state-dir are required'
);
if
(
args
.
imageFiles
&&
(
args
.
tick
||
args
.
apply
))
throw
new
Error
(
'--image-files is read-only and cannot combine with apply/tick'
);
if
(
args
.
backfillFileRefs
&&
(
args
.
tick
||
args
.
imageFiles
))
throw
new
Error
(
'--backfill-file-refs cannot combine with image-files/tick/watch'
);
if
(
args
.
resumeIdentity
&&
(
args
.
tick
||
args
.
imageFiles
||
args
.
backfillFileRefs
))
throw
new
Error
(
'--resume-identity cannot combine with other operations'
);
if
(
args
.
tick
&&
!
args
.
apply
)
throw
new
Error
(
'--tick/--watch requires --apply; preview cannot advance cursor'
);
return
args
;
}
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'
+
console
.
log
(
'meyer-imaging --config=<json> --state-dir=<private-directory> [--apply] [--tick | --watch] [--image-files=<imageId>] [--backfill-file-refs]
[--resume-identity=<imageId>]
\
n'
+
'Default: collect and validate only. --apply writes existing Friday patients through canonical fact parsers.
\
n'
+
'Required env: MEYER_APP_KEY, MEYER_APP_SECRET; DATABASE_URL is required for automatic organization mapping, including preview.
\
n'
+
'Default fixed window; --tick advances private cursor, --watch uses configured interval (default 15 minutes). No reconciliation, deletion, main scheduler, or recall recomputation.'
);
...
...
@@ -99,6 +105,20 @@ export async function runMeyerCli(argv: string[], isolateWatchFailures = false):
return
result
.
conflicts
||
result
.
raced
?
1
:
0
;
}
finally
{
await
db
.
$disconnect
();
}
}
if
(
args
.
resumeIdentity
)
{
const
db
=
new
PrismaClient
();
try
{
const
result
=
await
resumeIdentity
(
config
,
store
,
args
.
resumeIdentity
,
c
=>
client
.
forWindow
(
c
),
async
(
scope
,
image
)
=>
{
const
host
=
await
db
.
host
.
findUnique
({
where
:
{
name
:
'friday'
},
select
:
{
id
:
true
}
});
const
patient
=
host
&&
await
db
.
patient
.
findUnique
({
where
:
{
hostId_tenantId_sourceUnit_externalId
:
{
hostId
:
host
.
id
,
tenantId
:
scope
.
tenantId
,
sourceUnit
:
scope
.
sourceUnit
,
externalId
:
image
.
patientId
!
,
}
},
select
:
{
id
:
true
}
});
if
(
!
patient
)
throw
new
Error
(
'patient not present in configured Friday scope'
);
},
args
.
apply
);
console
.
log
(
JSON
.
stringify
({
mode
:
'resume-identity'
,
...
result
}));
return
0
;
}
finally
{
await
db
.
$disconnect
();
}
}
if
(
args
.
imageFiles
)
{
const
matches
:
Array
<
{
key
:
string
;
saved
:
SavedBundle
}
>
=
[];
for
(
const
scope
of
config
.
scopes
)
{
...
...
@@ -130,7 +150,7 @@ export async function runMeyerCli(argv: string[], isolateWatchFailures = false):
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
,
manualReview
:
status
.
manualReview
,
retiredUnsupported
:
status
.
retiredUnsupported
,
ret
iredIdentity
:
status
.
retiredIdentity
,
ret
ryAttempted
:
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
}));
...
...
apps/pac-service/src/integrations/meyer/README.md
View file @
ff1c0399
...
...
@@ -96,12 +96,25 @@ pnpm meyer-imaging -- --config=/absolute/meyer.json --state-dir=/absolute/privat
-
每家在分得的额度内,优先重试最早到期的待处理项。全景片从影像入库时刻起,前 15 分钟按正常轮次补查,之后每小时最多查一次。累计实际调用分析接口 8 次仍未取得结果,停止自动查询,转人工处理;记录和已入库影像保留。旧队列的混合重试次数不计入分析次数。
-
当前只支持全景片分析(aiType=3,兼容明确的全景片名称/类型)。其他类型只保存影像,不进入分析补查队列;类型不明时保留人工核实。分类记录写入 analysis-status,不能解释为分析阴性。升级时本地清理已落库且没有待写分析快照的旧非分析类型,不额外请求美亚、不占重试名额。
-
次数在请求前持久化,进程重启或重叠扫描不会重置。分析请求异常也计次数;已取得分析但写入失败的,使用原快照继续恢复写入,不受查询上限限制。身份校验、患者未绑定和影像写入失败另行退避,不消耗分析次数。
-
身份异常(来源 patientId 缺失、PAC 对应患者不存在或归属冲突)最多自动补查 8 次,首次采集失败不占补查次数。补查次数在来源请求前落盘,中断或请求异常也计本次补查。达到上限仍未恢复时原子移入 identity-review/,不再占活跃队列及网络额度;重叠扫描不会自动重新入队。数据库等一般写入故障不计为身份异常。旧缺 patientId 条目按原 attempts 减去首次采集折算,已经超限的保留原次数并直接转人工,不额外请求来源。
-
重试通过原影像入库时间前后 1 秒的窄窗重新定位 imageId,再获取分析。不会扫描整个诊所历史。身份改绑、影像不可见均保留待处理并报错,不自动撤销或迁移事实。
-
因接口没有按 imageId 直接查询元数据的能力,缺少入库时间的条目不能自动定位,须人工处理;这类情况会有 retryFailed。
-
scheduler-status.json 提供本轮各诊所结果、独立水位、sourceRequests、clinicFailures、组织映射异常 scopeFailures,以及
总待处理数 pending、自动补查数 autoPending、人工处理数 manualReview、当轮移出的不适用条数 retiredUnsupported 和重试失败数
。单家诊所故障按 15 分钟至最多 1 小时退避;单家组织映射缺失或冲突也单独记录,正常诊所继续。数据库不可用等共享依赖故障仍可能影响整轮;watch 仅对共享依赖异常额外退避,不因单诊所失败降低其他诊所频率;依赖恢复后继续,不丢弃失败数据。配置错误、损坏状态或归属冲突需要修复后才能恢复。SIGINT/SIGTERM 会等当前一轮收尾再停止。
-
scheduler-status.json 提供本轮各诊所结果、独立水位、sourceRequests、clinicFailures、组织映射异常 scopeFailures,以及
活跃目录待处理数 pending、自动补查数 autoPending、其中停止自动分析的 manualReview、当轮移出的不适用条数 retiredUnsupported、身份超限条数 retiredIdentity 和重试失败数。identity-review/ 不计入 pending/manualReview,也不在定时任务中枚举
。单家诊所故障按 15 分钟至最多 1 小时退避;单家组织映射缺失或冲突也单独记录,正常诊所继续。数据库不可用等共享依赖故障仍可能影响整轮;watch 仅对共享依赖异常额外退避,不因单诊所失败降低其他诊所频率;依赖恢复后继续,不丢弃失败数据。配置错误、损坏状态或归属冲突需要修复后才能恢复。SIGINT/SIGTERM 会等当前一轮收尾再停止。
-
已成功分析的影像不再重新查询分析;只有影像阶段完成且符合补查策略时才自动重试。这是首次接入及晚到分析重试,不是历史对账;删除、撤回和成功后的修改仍不覆盖。
-
状态目录必须持久化。先前仅有 deferred-
*
的旧目录没有完整队列,升级后先按原窗口重跑一次再启动 tick/watch。不要用曾做远期手工采集的目录直接推断历史已覆盖。
## 身份异常恢复
身份绑定修复后,使用同一配置和状态目录恢复单张影像。先预检,不改队列:
```
sh
pnpm meyer-imaging
--config
=
/private/config.json
--state-dir
=
/private/state
--resume-identity
=
<imageId>
# 预检通过后加 --apply,重新进入原定时任务;恢复命令本身不写患者事实。
pnpm meyer-imaging
--config
=
/private/config.json
--state-dir
=
/private/state
--resume-identity
=
<imageId>
--apply
```
恢复会重新读取源头影像,检查授权诊所、PAC 患者完整身份和已有快照归属;仍缺绑定、患者未同步或需要改绑已有事实时拒绝恢复。只清零本轮身份补查次数,保留分析次数及累计 attempts,原异常记录留在 identity-history/,不修改历史事实。与采集任务共用根状态锁,不能并发操作。
## 原图地址按需刷新
```
sh
...
...
@@ -177,7 +190,7 @@ clinics/<诊所哈希>/ 下的 cursor、scan-progress、scan-ledger、failure
### 存储分工
-
**独立状态目录**
:identity、数据库目标指纹、scheduler-layout、clinic-rotation、clinics/ 各诊所进度、pending、analysis-status、image、bundle、observed、scheduler-status、access 文件;本期不新增数据库采集表。状态目录须放持久磁盘、限制权限、备份。只支持单机单写者,不用于多机器共享抢任务。
-
**独立状态目录**
:identity、数据库目标指纹、scheduler-layout、clinic-rotation、clinics/ 各诊所进度、pending、
identity-review、identity-history、
analysis-status、image、bundle、observed、scheduler-status、access 文件;本期不新增数据库采集表。状态目录须放持久磁盘、限制权限、备份。只支持单机单写者,不用于多机器共享抢任务。
-
**PAC 共用表**
:标准化事件写 patient_transactions,标准影像/诊断写 patient_facts,按 Friday 宿主、租户、品牌与患者隔离;patients 只查找,不创建或移动。原始美亚依据还保留在事务 rawPayload。
-
**原片**
:仍在美亚/OSS,本地只留文件 ID 和业务证据,不复制文件。
...
...
apps/pac-service/src/integrations/meyer/collector.ts
View file @
ff1c0399
...
...
@@ -4,11 +4,15 @@ import { ImageSchema, SymptomSchema, hash, imageKey, sourceTime,
import
{
mapMeyerImage
,
mapMeyerAnalysis
}
from
'./mapper'
;
import
{
MeyerStore
,
type
SavedBundle
}
from
'./store'
;
import
{
analysisCapability
,
nextAnalysisRetry
}
from
'./retry-policy'
;
import
{
identityReviewStore
,
isIdentityPending
,
MeyerIdentityError
,
retireIdentity
}
from
'./identity-retry'
;
export
interface
PendingImage
{
scope
:
MeyerScope
;
image
:
MeyerImage
;
attempts
:
number
;
nextRetryAt
:
string
;
/** 只计实际分析请求;旧 attempts 同时包含身份/落库失败,不能拿来作为分析次数。 */
analysisAttempts
?:
number
;
autoRetry
?:
boolean
;
identityAttempts
?:
number
;
identityIssue
?:
string
;
identityReview
?:
{
exhaustedAt
:
string
;
limit
:
number
;
issue
:
string
};
identityRecovery
?:
{
resumedAt
:
string
;
historyKey
:
string
};
reason
?:
'analysis_pending'
|
'analysis_error'
|
'analysis_exhausted'
|
'analysis_type_unknown'
|
'image_or_identity'
|
'analysis_apply'
;
}
export
interface
CollectResult
{
...
...
@@ -18,7 +22,7 @@ export interface CollectResult {
/** 固定时间窗单次运行;不推进主流程游标,不删除、不改绑、不复查已完成影像。 */
export
async
function
collectMeyer
(
config
:
MeyerConfig
,
source
:
MeyerSource
,
store
:
MeyerStore
,
codes
:
Record
<
string
,
string
>
,
sink
?:
MeyerSink
,
respectRetryAt
=
false
,
now
=
Date
.
now
()):
Promise
<
CollectResult
>
{
codes
:
Record
<
string
,
string
>
,
sink
?:
MeyerSink
,
respectRetryAt
=
false
,
now
=
Date
.
now
()
,
identityRetryInFlight
=
false
):
Promise
<
CollectResult
>
{
const
pending
=
new
MeyerStore
(
join
(
store
.
directory
,
'pending'
));
const
result
:
CollectResult
=
{
pages
:
0
,
seen
:
0
,
collected
:
0
,
applied
:
0
,
alreadyApplied
:
0
,
deferred
:
0
,
failed
:
0
,
imagesApplied
:
0
,
analysisFailed
:
0
,
errors
:
[]
};
...
...
@@ -55,7 +59,14 @@ export async function collectMeyer(config: MeyerConfig, source: MeyerSource, sto
if
(
image
.
imageCreateTime
&&
(
sourceTime
(
image
.
imageCreateTime
)
<
sourceTime
(
config
.
timeStart
)
||
sourceTime
(
image
.
imageCreateTime
)
>
sourceTime
(
config
.
timeEnd
)))
throw
new
Error
(
'image outside requested window'
);
const
previousRetry
=
await
pending
.
read
<
PendingImage
>
(
key
);
// 超限条目只按键查看,不扫描人工目录;重叠窗口和手工固定窗口也不能绕过恢复入口。
if
(
await
identityReviewStore
(
store
).
read
(
key
))
{
result
.
deferred
++
;
continue
;
}
if
(
!
identityRetryInFlight
&&
previousRetry
&&
await
retireIdentity
(
store
,
key
,
previousRetry
,
now
))
{
result
.
deferred
++
;
continue
;
}
if
(
respectRetryAt
&&
previousRetry
&&
isIdentityPending
(
previousRetry
)
&&
Date
.
parse
(
previousRetry
.
nextRetryAt
)
>
now
)
{
result
.
deferred
++
;
continue
;
}
retry
=
{
scope
,
image
,
attempts
:
(
previousRetry
?.
attempts
??
0
)
+
1
,
nextRetryAt
:
''
,
identityAttempts
:
previousRetry
?.
identityAttempts
??
0
,
...(
previousRetry
?.
identityRecovery
?
{
identityRecovery
:
previousRetry
.
identityRecovery
}
:
{}),
analysisAttempts
:
previousRetry
?.
analysisAttempts
??
0
,
autoRetry
:
true
,
reason
:
'image_or_identity'
};
const
saved
=
await
store
.
read
<
SavedBundle
>
(
`bundle-
${
key
}
`
);
const
metadata
=
await
store
.
read
<
SavedBundle
>
(
`image-
${
key
}
`
);
...
...
@@ -69,7 +80,7 @@ export async function collectMeyer(config: MeyerConfig, source: MeyerSource, sto
throw
new
Error
(
'saved bundle scope mismatch'
);
}
if
(
retained
.
bundle
.
image
.
patientExternalId
!==
image
.
patientId
)
{
throw
new
Error
(
'image ownership changed; reconciliation is not supported'
);
throw
new
MeyerIdentity
Error
(
'image ownership changed; reconciliation is not supported'
);
}
}
if
(
saved
?.
applied
)
{
completed
=
true
;
result
.
alreadyApplied
++
;
continue
;
}
...
...
@@ -77,6 +88,7 @@ export async function collectMeyer(config: MeyerConfig, source: MeyerSource, sto
if
(
saved
?.
bundle
.
version
!==
1
)
{
const
imageBundle
=
metadata
?.
bundle
??
mapMeyerImage
(
config
,
scope
,
image
,
new
Date
().
toISOString
());
if
(
!
imageBundle
)
{
if
(
!
image
.
patientId
)
retry
.
identityIssue
=
'missing patientId'
;
await
store
.
write
(
`deferred-
${
key
}
`
,
{
reason
:
'unbound patient or missing image creation time'
,
image
});
result
.
deferred
++
;
continue
;
}
...
...
@@ -148,6 +160,7 @@ export async function collectMeyer(config: MeyerConfig, source: MeyerSource, sto
}
completed
=
!!
sink
;
}
catch
(
error
)
{
if
(
error
instanceof
MeyerIdentityError
&&
retry
)
retry
.
identityIssue
=
error
.
message
;
if
(
requestedAnalysis
&&
retry
?.
reason
!==
'analysis_apply'
)
retry
!
.
reason
=
'analysis_error'
;
if
(
imageReady
)
result
.
analysisFailed
++
;
else
result
.
failed
++
;
result
.
errors
.
push
({
key
,
reason
:
error
instanceof
Error
?
error
.
message
:
'collection failed'
});
...
...
@@ -162,6 +175,7 @@ export async function collectMeyer(config: MeyerConfig, source: MeyerSource, sto
}
}
else
retry
.
nextRetryAt
=
new
Date
(
now
+
Math
.
min
(
3600000
,
60000
*
2
**
Math
.
min
(
6
,
retry
.
attempts
-
1
))).
toISOString
();
await
pending
.
write
(
key
,
retry
);
await
retireIdentity
(
store
,
key
,
retry
,
now
);
}
}
}
...
...
apps/pac-service/src/integrations/meyer/identity-retry.ts
0 → 100644
View file @
ff1c0399
import
{
randomUUID
}
from
'node:crypto'
;
import
{
join
}
from
'node:path'
;
import
type
{
PendingImage
}
from
'./collector'
;
import
{
hash
,
imageKey
,
ImageSchema
,
sourceTime
,
type
MeyerConfig
,
type
MeyerImage
,
type
MeyerScope
,
type
MeyerSource
}
from
'./contracts'
;
import
{
MeyerStore
,
type
SavedBundle
}
from
'./store'
;
export
const
MAX_IDENTITY_RETRIES
=
8
;
export
class
MeyerIdentityError
extends
Error
{}
export
const
identityReviewStore
=
(
store
:
MeyerStore
)
=>
new
MeyerStore
(
join
(
store
.
directory
,
'identity-review'
));
export
const
isIdentityPending
=
(
p
:
PendingImage
)
=>
!!
p
.
identityIssue
||
(
p
.
reason
===
'image_or_identity'
&&
!
p
.
image
.
patientId
);
// 旧 attempts 混合多种失败;只有明确缺患者 ID 的旧条目可折算,首次采集不计补查次数。
export
function
identityRetries
(
p
:
PendingImage
)
{
const
count
=
p
.
identityAttempts
??
(
p
.
reason
===
'image_or_identity'
&&
!
p
.
image
.
patientId
?
Math
.
max
(
0
,
p
.
attempts
-
1
)
:
0
);
if
(
!
Number
.
isSafeInteger
(
count
)
||
count
<
0
)
throw
new
Error
(
'invalid identity retry count'
);
return
count
;
}
export
async
function
retireIdentity
(
store
:
MeyerStore
,
key
:
string
,
p
:
PendingImage
,
now
:
number
)
{
if
(
!
isIdentityPending
(
p
)
||
identityRetries
(
p
)
<
MAX_IDENTITY_RETRIES
)
return
false
;
if
(
key
!==
imageKey
(
p
.
scope
,
p
.
image
.
imageId
))
throw
new
Error
(
'identity queue key mismatch'
);
const
pending
=
new
MeyerStore
(
join
(
store
.
directory
,
'pending'
));
const
review
=
identityReviewStore
(
store
);
if
(
await
review
.
read
(
key
))
throw
new
Error
(
'identity review already exists; inspect retained state'
);
// 先持久化停止标记,再原子移出活跃目录;中断后下一轮只收尾迁移,不再请求来源。
await
pending
.
write
(
key
,
{
...
p
,
identityAttempts
:
identityRetries
(
p
),
autoRetry
:
false
,
identityReview
:
{
exhaustedAt
:
new
Date
(
now
).
toISOString
(),
limit
:
MAX_IDENTITY_RETRIES
,
issue
:
p
.
identityIssue
??
'missing patientId'
}
});
await
pending
.
moveTo
(
key
,
review
);
return
true
;
}
/** 人工恢复仅重新入队;原异常快照永久留底,后续仍走同一采集/事实写入流程。 */
export
async
function
resumeIdentity
(
config
:
MeyerConfig
,
store
:
MeyerStore
,
id
:
string
,
sourceFor
:
(
config
:
MeyerConfig
)
=>
MeyerSource
,
validatePatient
:
(
scope
:
MeyerScope
,
image
:
MeyerImage
)
=>
Promise
<
void
>
,
apply
:
boolean
,
now
=
Date
.
now
())
{
const
review
=
identityReviewStore
(
store
);
const
matches
:
Array
<
{
key
:
string
;
p
:
PendingImage
}
>
=
[];
for
(
const
scope
of
config
.
scopes
)
{
const
key
=
imageKey
(
scope
,
id
),
p
=
await
review
.
read
<
PendingImage
>
(
key
);
if
(
p
)
{
if
(
hash
(
p
.
scope
)
!==
hash
(
scope
)
||
p
.
image
.
imageId
!==
id
)
throw
new
Error
(
'identity review scope mismatch'
);
matches
.
push
({
key
,
p
});
}
}
if
(
matches
.
length
!==
1
)
throw
new
Error
(
'requires exactly one identity review in configured scope'
);
const
{
key
,
p
}
=
matches
[
0
]
!
;
if
(
!
p
.
image
.
imageCreateTime
)
throw
new
Error
(
'image creation time required for identity recovery'
);
const
t
=
Date
.
parse
(
sourceTime
(
p
.
image
.
imageCreateTime
));
const
china
=
(
ms
:
number
)
=>
new
Date
(
ms
+
8
*
3600000
).
toISOString
().
slice
(
0
,
19
).
replace
(
'T'
,
' '
);
const
client
=
sourceFor
({
...
config
,
timeStart
:
china
(
t
-
1000
),
timeEnd
:
china
(
t
+
1000
)
});
let
found
:
MeyerImage
|
undefined
,
total
:
number
|
undefined
;
for
(
let
page
=
1
;
;
page
++
)
{
if
(
page
>
Math
.
min
(
config
.
maxPages
,
config
.
maxListPagesPerTick
,
config
.
maxSourceRequestsPerTick
))
throw
new
Error
(
'identity recovery page limit'
);
const
batch
=
await
client
.
list
(
p
.
scope
,
page
);
if
(
batch
.
page
!==
page
||
batch
.
size
!==
config
.
pageSize
||
(
total
!==
undefined
&&
total
!==
batch
.
total
)
||
batch
.
images
.
length
!==
Math
.
min
(
batch
.
size
,
Math
.
max
(
0
,
batch
.
total
-
(
page
-
1
)
*
batch
.
size
)))
throw
new
Error
(
'identity recovery page changed'
);
total
=
batch
.
total
;
for
(
const
raw
of
batch
.
images
)
if
((
raw
as
{
imageId
?:
string
}).
imageId
===
id
)
{
if
(
found
)
throw
new
Error
(
'duplicate identity recovery image'
);
found
=
ImageSchema
.
parse
(
raw
);
}
if
(
page
*
batch
.
size
>=
total
)
break
;
}
if
(
!
found
||
!
found
.
patientId
||
found
.
clinicId
!==
p
.
scope
.
clinicId
||
found
.
imageCreateTime
!==
p
.
image
.
imageCreateTime
)
throw
new
Error
(
'source image identity not ready'
);
// 已经留底的事实不能靠恢复操作改绑;无成功写入时也保守保留原 canonical 快照。
for
(
const
prefix
of
[
'image'
,
'bundle'
])
{
const
saved
=
await
store
.
read
<
SavedBundle
>
(
`
${
prefix
}
-
${
key
}
`
);
if
(
saved
&&
(
hash
(
saved
.
bundle
.
scope
)
!==
hash
(
p
.
scope
)
||
saved
.
bundle
.
namespace
!==
config
.
namespace
||
saved
.
bundle
.
raw
.
image
.
imageId
!==
id
||
saved
.
bundle
.
image
.
patientExternalId
!==
found
.
patientId
))
{
throw
new
Error
(
'saved image ownership differs; reconciliation required'
);
}
}
await
validatePatient
(
p
.
scope
,
found
);
if
(
!
apply
)
return
{
ready
:
true
,
resumed
:
false
};
const
pending
=
new
MeyerStore
(
join
(
store
.
directory
,
'pending'
));
if
(
await
pending
.
read
(
key
))
throw
new
Error
(
'active identity entry already exists'
);
// 原文先留不可变历史,再写恢复状态并原子移回;任何中断都不会丢失原数据。
const
history
=
new
MeyerStore
(
join
(
store
.
directory
,
'identity-history'
,
key
));
const
historyKey
=
randomUUID
();
await
history
.
write
(
historyKey
,
p
);
const
{
identityReview
:
_review
,
identityIssue
:
_issue
,
...
previous
}
=
p
;
await
review
.
write
(
key
,
{
...
previous
,
image
:
found
,
identityAttempts
:
0
,
autoRetry
:
true
,
nextRetryAt
:
new
Date
(
now
).
toISOString
(),
reason
:
'image_or_identity'
,
identityRecovery
:
{
resumedAt
:
new
Date
(
now
).
toISOString
(),
historyKey
}
});
await
review
.
moveTo
(
key
,
pending
);
return
{
ready
:
true
,
resumed
:
true
};
}
apps/pac-service/src/integrations/meyer/pac-sink.ts
View file @
ff1c0399
...
...
@@ -5,6 +5,7 @@ import { ParserPipeline } from '../../modules/sync/pipeline/parser-pipeline.serv
import
{
TransactionSynthesizer
}
from
'../../modules/sync/pipeline/transaction-synthesizer'
;
import
{
hash
,
type
MeyerBundle
,
type
MeyerSink
}
from
'./contracts'
;
import
{
mapMeyerFileRefs
,
projectMeyerImageFiles
}
from
'./file-refs'
;
import
{
MeyerIdentityError
}
from
'./identity-retry'
;
/** 标准字段边界之后才调用已有事实内核;不启动 Push/Pull、主档写入或召回任务。 */
@
Injectable
()
...
...
@@ -32,7 +33,7 @@ export class MeyerPacSink implements MeyerSink {
externalId
:
image
.
patientExternalId
},
},
select
:
{
id
:
true
}
});
// 补充路线只补既有宿主患者,不凭美亚数据创建或移动患者。
if
(
!
patient
)
throw
new
Error
(
'patient not present in configured Friday scope'
);
if
(
!
patient
)
throw
new
MeyerIdentity
Error
(
'patient not present in configured Friday scope'
);
const
rows
=
[{
resource
:
'image_study'
,
row
:
image
,
action
:
Action
.
IMAGE_UPLOADED
,
subjectType
:
SubjectType
.
IMAGE_STUDY
,
timeField
:
'uploadedAt'
},
...
diagnoses
.
map
(
row
=>
({
resource
:
'diagnosis'
,
row
,
action
:
Action
.
DIAGNOSIS_RECORDED
,
...
...
@@ -44,7 +45,7 @@ export class MeyerPacSink implements MeyerSink {
hostId
:
host
.
id
,
tenantId
,
sourceUnit
,
subjectId
:
`image_record:
${
image
.
externalId
}
`
,
},
orderBy
:
{
version
:
'desc'
}
});
if
(
current
&&
(
current
.
patientId
!==
patient
.
id
||
current
.
clinicId
!==
clinicId
))
{
throw
new
Error
(
'existing supplemental snapshot differs; image ownership changed'
);
throw
new
MeyerIdentity
Error
(
'existing supplemental snapshot differs; image ownership changed'
);
}
if
((
current
?.
content
as
{
analysis
?:
unknown
}
|
undefined
)?.
analysis
)
continue
;
}
...
...
@@ -66,8 +67,10 @@ export class MeyerPacSink implements MeyerSink {
hostId
:
host
.
id
,
tenantId
,
sourceEventId
:
data
.
sourceEventId
,
}
});
const
previous
=
transaction
.
rawPayload
as
{
_meyer
?:
{
canonicalHash
?:
string
;
namespace
?:
string
;
image
?:
MeyerBundle
[
'raw'
][
'image'
]
}
};
if
(
transaction
.
patientId
!==
patient
.
id
||
transaction
.
clinicId
!==
clinicId
||
previous
?.
_meyer
?.
canonicalHash
!==
canonicalHash
)
throw
new
Error
(
'existing supplemental snapshot differs; reconciliation is not supported'
);
if
(
transaction
.
patientId
!==
patient
.
id
||
transaction
.
clinicId
!==
clinicId
)
{
throw
new
MeyerIdentityError
(
'existing supplemental snapshot differs; image ownership changed'
);
}
if
(
previous
?.
_meyer
?.
canonicalHash
!==
canonicalHash
)
throw
new
Error
(
'existing supplemental snapshot differs; reconciliation is not supported'
);
if
(
item
.
resource
===
'image_study'
&&
(
!
previous
.
_meyer
?.
image
||
previous
.
_meyer
.
namespace
!==
bundle
.
namespace
||
hash
(
mapMeyerFileRefs
(
bundle
.
namespace
,
previous
.
_meyer
.
image
))
!==
hash
(
projectedImage
.
fileRefs
??
[])))
{
throw
new
Error
(
'existing supplemental snapshot differs; file references changed'
);
...
...
apps/pac-service/src/integrations/meyer/scheduler.ts
View file @
ff1c0399
...
...
@@ -7,6 +7,7 @@ import { imageKey, sourceTime, type MeyerConfig, type MeyerSource, type MeyerSin
import
{
analysisCapability
}
from
'./retry-policy'
;
import
{
clinicStateStore
,
prepareClinicStates
}
from
'./clinic-state'
;
import
{
MeyerBudgetExhausted
,
MeyerTickBudget
}
from
'./tick-budget'
;
import
{
identityReviewStore
,
identityRetries
,
isIdentityPending
,
MeyerIdentityError
,
retireIdentity
}
from
'./identity-retry'
;
export
const
chinaTime
=
(
ms
:
number
)
=>
new
Date
(
ms
+
8
*
3600000
).
toISOString
().
slice
(
0
,
19
).
replace
(
'T'
,
' '
);
/** 一轮独立任务:先限量重试,再扫描一个窗口;只推进本适配器水位。 */
...
...
@@ -21,7 +22,7 @@ async function runClinicTick(config: MeyerConfig, sourceFor: (c: MeyerConfig) =>
const
end
=
Math
.
min
(
now
-
60000
,
through
+
3600000
);
const
pending
=
new
MeyerStore
(
join
(
store
.
directory
,
'pending'
));
const
due
:
Array
<
{
key
:
string
;
value
:
PendingImage
}
>
=
[];
let
retiredUnsupported
=
0
;
let
retiredUnsupported
=
0
,
retiredIdentity
=
0
;
for
(
const
key
of
pendingKeys
)
{
const
value
=
await
pending
.
read
<
PendingImage
>
(
key
);
if
(
!
value
)
continue
;
...
...
@@ -45,7 +46,9 @@ async function runClinicTick(config: MeyerConfig, sourceFor: (c: MeyerConfig) =>
}
due
.
sort
((
a
,
b
)
=>
a
.
value
.
nextRetryAt
.
localeCompare
(
b
.
value
.
nextRetryAt
));
let
retryFailed
=
0
,
retryAttempted
=
0
;
for
(
const
{
key
,
value
}
of
due
.
slice
(
0
,
config
.
maxRetriesPerTick
))
{
for
(
const
entry
of
due
.
slice
(
0
,
config
.
maxRetriesPerTick
))
{
const
{
key
}
=
entry
;
let
value
=
entry
.
value
;
if
(
!
available
())
break
;
retryAttempted
++
;
if
(
hash
(
value
.
scope
)
!==
hash
(
config
.
scopes
[
0
]))
throw
new
Error
(
'pending scope mismatch'
);
...
...
@@ -55,6 +58,11 @@ async function runClinicTick(config: MeyerConfig, sourceFor: (c: MeyerConfig) =>
const
client
=
sourceFor
(
retryConfig
);
// 每次重试重新读取影像身份,避免拿旧 patientId 给后来改绑的影像写事实。
try
{
if
(
isIdentityPending
(
value
))
{
value
=
{
...
value
,
identityAttempts
:
identityRetries
(
value
)
+
1
};
// 补查次数先持久化;来源请求中断也不能在重启后重新取得次数额度。
await
pending
.
write
(
key
,
value
);
}
let
found
:
unknown
;
let
total
:
number
|
undefined
;
for
(
let
page
=
1
;
;
page
++
)
{
...
...
@@ -70,21 +78,28 @@ async function runClinicTick(config: MeyerConfig, sourceFor: (c: MeyerConfig) =>
}
if
(
!
found
)
throw
new
Error
(
'pending image no longer visible'
);
if
(
value
.
image
.
patientId
&&
(
found
as
{
patientId
?:
string
}).
patientId
!==
value
.
image
.
patientId
)
{
throw
new
Error
(
'pending image ownership changed'
);
throw
new
MeyerIdentity
Error
(
'pending image ownership changed'
);
}
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
);
},
store
,
codes
,
sink
,
true
,
now
,
true
);
retryFailed
+=
result
.
failed
+
result
.
analysisFailed
;
if
(
await
identityReviewStore
(
store
).
read
(
key
))
retiredIdentity
++
;
}
catch
(
error
)
{
if
(
error
instanceof
MeyerBudgetExhausted
)
break
;
retryFailed
++
;
// 条目可能已移出队列,随后写摘要或 sync 失败;不能用旧值把人工记录重新激活。
if
(
await
identityReviewStore
(
store
).
read
(
key
))
{
retiredIdentity
++
;
continue
;
}
// collector 可能已持久化请求次数后才在写运行摘要时失败;不能回写旧次数。
const
retained
=
await
pending
.
read
<
PendingImage
>
(
key
)
??
value
;
await
pending
.
write
(
key
,
{
...
retained
,
attempts
:
retained
.
attempts
+
1
,
nextRetryAt
:
new
Date
(
now
+
3600000
).
toISOString
()
});
const
next
=
{
...
retained
,
attempts
:
retained
.
attempts
+
1
,
...(
error
instanceof
MeyerIdentityError
?
{
identityIssue
:
error
.
message
,
identityAttempts
:
retained
.
identityAttempts
??
1
}
:
{}),
nextRetryAt
:
new
Date
(
now
+
3600000
).
toISOString
()
};
await
pending
.
write
(
key
,
next
);
if
(
await
retireIdentity
(
store
,
key
,
next
,
now
))
retiredIdentity
++
;
}
}
let
scan
:
Awaited
<
ReturnType
<
typeof
scanWindow
>>
|
undefined
;
...
...
@@ -111,7 +126,7 @@ async function runClinicTick(config: MeyerConfig, sourceFor: (c: MeyerConfig) =>
// 新扫描产生的队列项在整轮汇总时再统一统计。
for
(
const
key
of
keys
)
if
((
await
pending
.
read
<
PendingImage
>
(
key
))?.
autoRetry
===
false
)
manualReview
++
;
const
status
=
{
checkedAt
:
new
Date
(
now
).
toISOString
(),
retryAttempted
,
retryFailed
,
pending
:
keys
.
length
,
manualReview
,
autoPending
:
keys
.
length
-
manualReview
,
retiredUnsupported
,
pending
:
keys
.
length
,
manualReview
,
autoPending
:
keys
.
length
-
manualReview
,
retiredUnsupported
,
retiredIdentity
,
scan
,
cursor
:
await
progressStore
.
read
<
{
through
:
string
}
>
(
'cursor'
)
};
await
progressStore
.
write
(
'scheduler-status'
,
status
);
return
status
;
...
...
@@ -130,14 +145,18 @@ export async function runMeyerTick(config: MeyerConfig, sourceFor: (c: MeyerConf
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
activeScopes
=
new
Set
(
scopes
.
map
(
hash
));
const
pending
=
new
MeyerStore
(
join
(
store
.
directory
,
'pending'
));
const
grouped
=
new
Map
<
string
,
string
[]
>
();
let
queueErrors
=
0
;
let
queueErrors
=
0
,
retiredIdentity
=
0
;
for
(
const
key
of
await
pending
.
keys
())
{
try
{
const
value
=
await
pending
.
read
<
PendingImage
>
(
key
);
if
(
!
value
)
continue
;
const
scopeKey
=
hash
(
value
.
scope
);
if
(
activeScopes
.
has
(
scopeKey
)
&&
await
retireIdentity
(
store
,
key
,
value
,
now
))
{
retiredIdentity
++
;
continue
;
}
const
group
=
grouped
.
get
(
scopeKey
)
??
[];
group
.
push
(
key
);
grouped
.
set
(
scopeKey
,
group
);
}
catch
{
queueErrors
++
;
}
// 单个损坏条目留底,其他诊所仍可运行。
...
...
@@ -179,6 +198,7 @@ export async function runMeyerTick(config: MeyerConfig, sourceFor: (c: MeyerConf
retriesLeft
-=
result
.
retryAttempted
;
retryAttempted
+=
result
.
retryAttempted
;
retryFailed
+=
result
.
retryFailed
;
retiredUnsupported
+=
result
.
retiredUnsupported
;
retiredIdentity
+=
result
.
retiredIdentity
;
if
(
result
.
scan
)
{
pagesLeft
-=
result
.
scan
.
pages
;
imagesLeft
-=
result
.
scan
.
examined
;
for
(
const
key
of
[
'pages'
,
'seen'
,
'collected'
,
'applied'
,
'alreadyApplied'
,
'deferred'
,
'failed'
,
...
...
@@ -202,7 +222,6 @@ export async function runMeyerTick(config: MeyerConfig, sourceFor: (c: MeyerConf
}
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
);
...
...
@@ -225,7 +244,7 @@ export async function runMeyerTick(config: MeyerConfig, sourceFor: (c: MeyerConf
}
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
,
retryAttempted
,
retryFailed
,
retiredUnsupported
,
retiredIdentity
,
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
);
...
...
apps/pac-service/src/integrations/meyer/store.ts
View file @
ff1c0399
...
...
@@ -62,6 +62,15 @@ export class MeyerStore {
try
{
await
directory
.
sync
();
}
finally
{
await
directory
.
close
();
}
}
async
remove
(
key
:
string
):
Promise
<
void
>
{
await
rm
(
this
.
path
(
key
),
{
force
:
true
});
}
/** 调用方须持有根状态锁;队列转换不复制删除,保证条目只归属一个目录。 */
async
moveTo
(
key
:
string
,
target
:
MeyerStore
):
Promise
<
void
>
{
await
mkdir
(
target
.
directory
,
{
recursive
:
true
,
mode
:
0o700
});
await
rename
(
this
.
path
(
key
),
target
.
path
(
key
));
for
(
const
path
of
[
this
.
directory
,
target
.
directory
])
{
const
dir
=
await
open
(
path
,
'r'
);
try
{
await
dir
.
sync
();
}
finally
{
await
dir
.
close
();
}
}
}
async
keys
():
Promise
<
string
[]
>
{
try
{
return
(
await
readdir
(
this
.
directory
)).
filter
(
n
=>
/^
[
a-zA-Z0-9_-
]
+
\.
json$/
.
test
(
n
)).
map
(
n
=>
n
.
slice
(
0
,
-
5
));
}
catch
(
e
)
{
if
((
e
as
NodeJS
.
ErrnoException
).
code
===
'ENOENT'
)
return
[];
throw
e
;
}
...
...
apps/pac-service/tests/meyer-identity-retry.spec.ts
0 → 100644
View file @
ff1c0399
import
{
mkdtemp
,
rm
}
from
'node:fs/promises'
;
import
{
join
}
from
'node:path'
;
import
{
tmpdir
}
from
'node:os'
;
import
{
collectMeyer
,
type
PendingImage
}
from
'../src/integrations/meyer/collector'
;
import
{
hash
,
imageKey
,
type
MeyerImage
,
type
MeyerSource
}
from
'../src/integrations/meyer/contracts'
;
import
{
identityReviewStore
,
MeyerIdentityError
,
resumeIdentity
,
retireIdentity
}
from
'../src/integrations/meyer/identity-retry'
;
import
{
MeyerStore
}
from
'../src/integrations/meyer/store'
;
import
{
runMeyerTick
,
chinaTime
}
from
'../src/integrations/meyer/scheduler'
;
import
{
meyerConfig
,
meyerImage
,
meyerSymptom
,
codes
}
from
'./meyer-imaging-fixtures'
;
import
{
parseMeyerArgs
}
from
'../src/cli/meyer-imaging.cli'
;
describe
(
'Meyer bounded identity retries and recovery'
,
()
=>
{
let
dir
:
string
,
store
:
MeyerStore
;
const
now
=
Date
.
parse
(
'2026-10-17T03:00:00Z'
);
const
c
=
meyerConfig
(),
scope
=
c
.
scopes
[
0
]
!
;
const
missing
=
{
...
meyerImage
(),
patientId
:
null
,
medicalRecordNo
:
'INVALID-test'
};
const
key
=
imageKey
(
scope
,
missing
.
imageId
);
const
pending
=
()
=>
new
MeyerStore
(
join
(
dir
,
'pending'
));
const
review
=
()
=>
identityReviewStore
(
store
);
let
image
:
MeyerImage
;
let
source
:
MeyerSource
;
let
sink
:
{
apply
:
jest
.
Mock
<
Promise
<
void
>
,
any
[]
>
};
beforeEach
(
async
()
=>
{
dir
=
await
mkdtemp
(
join
(
tmpdir
(),
'meyer-identity-'
));
store
=
new
MeyerStore
(
dir
);
image
=
missing
;
source
=
{
list
:
jest
.
fn
(
async
(
_
,
page
)
=>
({
total
:
1
,
page
,
size
:
100
,
images
:
[
image
]
})),
symptom
:
jest
.
fn
(
async
()
=>
meyerSymptom
())
};
sink
=
{
apply
:
jest
.
fn
(
async
()
=>
{})
};
// 将扫描水位固定在测试补查时间之后,只测试真实补查请求,不触发额外扫描。
await
store
.
write
(
'cursor'
,
{
through
:
chinaTime
(
now
+
48
*
3600000
)
});
});
afterEach
(
async
()
=>
{
await
rm
(
dir
,
{
recursive
:
true
,
force
:
true
});
});
const
seed
=
async
(
overrides
:
Partial
<
PendingImage
>
=
{})
=>
{
const
p
:
PendingImage
=
{
scope
,
image
:
missing
,
attempts
:
1
,
identityAttempts
:
0
,
identityIssue
:
'missing patientId'
,
analysisAttempts
:
0
,
reason
:
'image_or_identity'
,
nextRetryAt
:
new
Date
(
now
-
3600000
).
toISOString
(),
...
overrides
};
await
pending
().
write
(
key
,
p
);
return
p
;
};
const
tick
=
(
n
=
0
)
=>
runMeyerTick
(
c
,
()
=>
source
,
store
,
codes
,
sink
,
now
+
n
*
3600000
);
test
(
'eight automatic identity retries, persisted before HTTP; archive is outside active enumeration'
,
async
()
=>
{
await
collectMeyer
(
c
,
source
,
store
,
codes
,
sink
,
true
,
now
-
3600000
);
expect
(
await
pending
().
read
(
key
)).
toMatchObject
({
identityAttempts
:
0
,
analysisAttempts
:
0
});
(
source
.
list
as
jest
.
Mock
).
mockClear
();
(
source
.
list
as
jest
.
Mock
).
mockImplementation
(
async
(
_s
,
page
)
=>
{
const
retained
=
await
pending
().
read
<
PendingImage
>
(
key
);
expect
(
retained
!
.
identityAttempts
).
toBeGreaterThan
(
0
);
return
{
total
:
1
,
page
,
size
:
100
,
images
:
[
image
]
};
});
for
(
let
n
=
0
;
n
<
8
;
n
++
)
await
tick
(
n
);
expect
(
source
.
list
).
toHaveBeenCalledTimes
(
8
);
expect
(
await
pending
().
keys
()).
toEqual
([]);
expect
(
await
review
().
read
(
key
)).
toMatchObject
({
image
:
missing
,
attempts
:
9
,
identityAttempts
:
8
,
analysisAttempts
:
0
,
autoRetry
:
false
,
identityReview
:
{
limit
:
8
}
});
await
tick
(
9
);
expect
(
source
.
list
).
toHaveBeenCalledTimes
(
8
);
expect
(
source
.
symptom
).
not
.
toHaveBeenCalled
();
expect
(
sink
.
apply
).
not
.
toHaveBeenCalled
();
});
test
(
'a repaired binding on the eighth request is processed before deciding exhaustion'
,
async
()
=>
{
await
seed
({
identityAttempts
:
7
});
image
=
meyerImage
();
const
result
=
await
tick
();
expect
(
result
.
retryFailed
).
toBe
(
0
);
expect
(
await
review
().
read
(
key
)).
toBeNull
();
expect
(
await
pending
().
read
(
key
)).
toBeNull
();
expect
(
await
store
.
read
(
`image-
${
key
}
`
)).
toMatchObject
({
applied
:
true
});
expect
(
await
store
.
read
(
`bundle-
${
key
}
`
)).
toMatchObject
({
applied
:
true
});
});
test
(
'legacy unbound attempts migrate locally without another source request; other failures are not inferred'
,
async
()
=>
{
await
seed
({
attempts
:
21
,
identityAttempts
:
undefined
,
identityIssue
:
undefined
});
const
result
=
await
tick
();
expect
(
result
).
toMatchObject
({
retiredIdentity
:
1
,
retryAttempted
:
0
,
pending
:
0
});
expect
(
source
.
list
).
not
.
toHaveBeenCalled
();
expect
(
await
review
().
read
(
key
)).
toMatchObject
({
attempts
:
21
,
identityAttempts
:
20
,
image
:
missing
});
});
test
(
'PAC patient absence uses the same cap; database outages retain normal write recovery'
,
async
()
=>
{
image
=
meyerImage
();
sink
.
apply
.
mockRejectedValue
(
new
MeyerIdentityError
(
'patient not present in configured Friday scope'
));
await
collectMeyer
(
c
,
source
,
store
,
codes
,
sink
,
true
,
now
-
3600000
);
for
(
let
n
=
0
;
n
<
8
;
n
++
)
await
tick
(
n
);
expect
(
await
review
().
read
(
key
)).
toMatchObject
({
identityAttempts
:
8
,
identityIssue
:
'patient not present in configured Friday scope'
});
expect
(
source
.
symptom
).
not
.
toHaveBeenCalled
();
const
other
=
{
...
meyerImage
(),
imageId
:
'db-outage'
},
otherKey
=
imageKey
(
scope
,
other
.
imageId
);
image
=
other
;
sink
.
apply
.
mockRejectedValue
(
Error
(
'database unavailable'
));
await
pending
().
write
(
otherKey
,
{
scope
,
image
:
other
,
attempts
:
21
,
nextRetryAt
:
new
Date
(
now
).
toISOString
(),
reason
:
'image_or_identity'
});
await
collectMeyer
(
c
,
source
,
store
,
codes
,
sink
,
true
,
now
);
expect
(
await
review
().
read
(
otherKey
)).
toBeNull
();
expect
(
await
pending
().
read
(
otherKey
)).
toMatchObject
({
attempts
:
22
});
});
test
(
'overlapping scans and fixed-window manual collection cannot silently reactivate archived identities'
,
async
()
=>
{
const
p
=
await
seed
({
identityAttempts
:
8
});
await
retireIdentity
(
store
,
key
,
p
,
now
);
const
fingerprint
=
hash
(
await
review
().
read
(
key
));
image
=
meyerImage
();
for
(
const
automatic
of
[
true
,
false
])
await
collectMeyer
(
c
,
source
,
store
,
codes
,
sink
,
automatic
,
now
);
expect
(
hash
(
await
review
().
read
(
key
))).
toBe
(
fingerprint
);
expect
(
await
pending
().
keys
()).
toEqual
([]);
expect
(
sink
.
apply
).
not
.
toHaveBeenCalled
();
});
test
(
'retiring an exhausted row does not block a healthy image later on the same page'
,
async
()
=>
{
await
seed
({
identityAttempts
:
8
});
const
healthy
=
{
...
meyerImage
(),
imageId
:
'healthy'
};
source
.
list
=
jest
.
fn
(
async
()
=>
({
total
:
2
,
page
:
1
,
size
:
100
,
images
:
[
missing
,
healthy
]
}));
const
result
=
await
collectMeyer
(
c
,
source
,
store
,
codes
,
sink
,
true
,
now
);
expect
(
result
).
toMatchObject
({
failed
:
0
,
imagesApplied
:
1
,
applied
:
1
,
deferred
:
1
});
expect
(
await
review
().
read
(
key
)).
not
.
toBeNull
();
expect
(
await
store
.
read
(
`image-
${
imageKey
(
scope
,
healthy
.
imageId
)}
`
)).
toMatchObject
({
applied
:
true
});
});
test
(
'recovery rejects unresolved source and wrong PAC scope without changing retained evidence'
,
async
()
=>
{
const
p
=
await
seed
({
identityAttempts
:
8
});
await
retireIdentity
(
store
,
key
,
p
,
now
);
const
original
=
hash
(
await
review
().
read
(
key
));
const
validate
=
jest
.
fn
(
async
()
=>
{});
await
expect
(
resumeIdentity
(
c
,
store
,
image
.
imageId
,
()
=>
source
,
validate
,
true
,
now
)).
rejects
.
toThrow
(
'not ready'
);
expect
(
validate
).
not
.
toHaveBeenCalled
();
expect
(
hash
(
await
review
().
read
(
key
))).
toBe
(
original
);
image
=
meyerImage
();
validate
.
mockRejectedValue
(
Error
(
'patient not present'
));
await
expect
(
resumeIdentity
(
c
,
store
,
image
.
imageId
,
()
=>
source
,
validate
,
true
,
now
)).
rejects
.
toThrow
(
'patient not present'
);
expect
(
hash
(
await
review
().
read
(
key
))).
toBe
(
original
);
expect
(
await
pending
().
keys
()).
toEqual
([]);
});
test
(
'preview is read-only; recovery keeps immutable source history and uses the normal fact pipeline'
,
async
()
=>
{
const
p
=
await
seed
({
identityAttempts
:
8
,
analysisAttempts
:
3
});
await
retireIdentity
(
store
,
key
,
p
,
now
);
const
archived
=
await
review
().
read
<
PendingImage
>
(
key
);
image
=
meyerImage
();
const
validate
=
jest
.
fn
(
async
()
=>
{});
expect
(
await
resumeIdentity
(
c
,
store
,
image
.
imageId
,
()
=>
source
,
validate
,
false
,
now
)).
toEqual
({
ready
:
true
,
resumed
:
false
});
expect
(
await
review
().
read
(
key
)).
toEqual
(
archived
);
expect
(
await
pending
().
keys
()).
toEqual
([]);
await
resumeIdentity
(
c
,
store
,
image
.
imageId
,
()
=>
source
,
validate
,
true
,
now
);
const
restored
=
await
pending
().
read
<
PendingImage
>
(
key
);
expect
(
restored
).
toMatchObject
({
identityAttempts
:
0
,
analysisAttempts
:
3
,
attempts
:
1
,
image
:
meyerImage
()
});
expect
(
await
review
().
read
(
key
)).
toBeNull
();
const
history
=
new
MeyerStore
(
join
(
dir
,
'identity-history'
,
key
));
expect
(
await
history
.
read
(
restored
!
.
identityRecovery
!
.
historyKey
)).
toEqual
(
archived
);
await
tick
();
expect
(
await
pending
().
read
(
key
)).
toBeNull
();
expect
(
await
store
.
read
(
`image-
${
key
}
`
)).
toMatchObject
({
applied
:
true
});
expect
(
await
store
.
read
(
`bundle-
${
key
}
`
)).
toMatchObject
({
applied
:
true
});
expect
(
await
history
.
read
(
restored
!
.
identityRecovery
!
.
historyKey
)).
toEqual
(
archived
);
});
test
(
'recovery cannot reassign an already retained canonical image to a different patient'
,
async
()
=>
{
const
p
=
await
seed
({
identityAttempts
:
8
});
await
retireIdentity
(
store
,
key
,
p
,
now
);
image
=
meyerImage
();
await
store
.
write
(
`image-
${
key
}
`
,
{
applied
:
true
,
bundle
:
{
namespace
:
c
.
namespace
,
scope
,
raw
:
{
image
:
meyerImage
()
},
image
:
{
patientExternalId
:
'different-patient'
}
}
});
await
expect
(
resumeIdentity
(
c
,
store
,
image
.
imageId
,
()
=>
source
,
async
()
=>
{},
true
,
now
)).
rejects
.
toThrow
(
'reconciliation'
);
expect
(
await
pending
().
keys
()).
toEqual
([]);
expect
(
await
review
().
read
(
key
)).
not
.
toBeNull
();
});
test
(
'interruption while moving an exhausted entry is finished locally on restart'
,
async
()
=>
{
await
seed
({
identityAttempts
:
8
});
const
move
=
jest
.
spyOn
(
MeyerStore
.
prototype
,
'moveTo'
).
mockRejectedValueOnce
(
Error
(
'interrupted'
));
const
failed
=
await
tick
();
move
.
mockRestore
();
expect
(
failed
.
queueErrors
).
toBe
(
1
);
expect
(
await
pending
().
read
(
key
)).
toMatchObject
({
identityAttempts
:
8
,
autoRetry
:
false
});
const
resumed
=
await
tick
(
1
);
expect
(
resumed
.
retiredIdentity
).
toBe
(
1
);
expect
(
await
pending
().
keys
()).
toEqual
([]);
expect
(
source
.
list
).
not
.
toHaveBeenCalled
();
});
test
(
'a summary failure after archiving cannot resurrect an exhausted pending entry'
,
async
()
=>
{
await
seed
({
identityAttempts
:
7
});
const
write
=
store
.
write
.
bind
(
store
);
const
fail
=
jest
.
spyOn
(
store
,
'write'
).
mockImplementation
(
async
(
k
,
v
)
=>
{
if
(
k
===
'last-run'
)
throw
Error
(
'summary unavailable'
);
return
write
(
k
,
v
);
});
await
tick
();
fail
.
mockRestore
();
expect
(
await
pending
().
keys
()).
toEqual
([]);
expect
(
await
review
().
read
(
key
)).
toMatchObject
({
identityAttempts
:
8
});
(
source
.
list
as
jest
.
Mock
).
mockClear
();
await
tick
(
1
);
expect
(
source
.
list
).
not
.
toHaveBeenCalled
();
});
test
(
'interrupted manual recovery remains inactive and preserves the original history'
,
async
()
=>
{
const
p
=
await
seed
({
identityAttempts
:
8
});
await
retireIdentity
(
store
,
key
,
p
,
now
);
const
original
=
await
review
().
read
(
key
);
image
=
meyerImage
();
const
move
=
jest
.
spyOn
(
MeyerStore
.
prototype
,
'moveTo'
).
mockRejectedValueOnce
(
Error
(
'interrupted'
));
await
expect
(
resumeIdentity
(
c
,
store
,
image
.
imageId
,
()
=>
source
,
async
()
=>
{},
true
,
now
)).
rejects
.
toThrow
(
'interrupted'
);
move
.
mockRestore
();
expect
(
await
pending
().
keys
()).
toEqual
([]);
const
history
=
new
MeyerStore
(
join
(
dir
,
'identity-history'
,
key
));
expect
(
await
history
.
read
((
await
history
.
keys
())[
0
]
!
)).
toEqual
(
original
);
await
tick
();
expect
(
sink
.
apply
).
not
.
toHaveBeenCalled
();
await
resumeIdentity
(
c
,
store
,
image
.
imageId
,
()
=>
source
,
async
()
=>
{},
true
,
now
);
expect
(
await
pending
().
read
(
key
)).
toMatchObject
({
identityAttempts
:
0
});
});
test
(
'recovery is exclusive with other CLI operations and requires a nonempty image ID'
,
()
=>
{
const
base
=
[
'--config=x'
,
'--state-dir=y'
];
expect
(
parseMeyerArgs
([...
base
,
'--resume-identity=image-1'
,
'--apply'
]).
resumeIdentity
).
toBe
(
'image-1'
);
for
(
const
other
of
[
'--tick'
,
'--watch'
,
'--backfill-file-refs'
,
'--image-files=image-1'
])
{
expect
(()
=>
parseMeyerArgs
([...
base
,
'--resume-identity=image-1'
,
'--apply'
,
other
])).
toThrow
();
}
expect
(()
=>
parseMeyerArgs
([...
base
,
'--resume-identity='
])).
toThrow
();
});
});
apps/pac-service/tests/meyer-imaging-postgres.spec.ts
View file @
ff1c0399
...
...
@@ -21,6 +21,7 @@ import { meyerConfig, meyerImage, meyerSymptom } from './meyer-imaging-fixtures'
import
{
runMeyerCli
}
from
'../src/cli/meyer-imaging.cli'
;
import
{
clinicStateStore
}
from
'../src/integrations/meyer/clinic-state'
;
import
{
backfillMeyerFileRefs
}
from
'../src/integrations/meyer/backfill-file-refs'
;
import
{
identityReviewStore
,
retireIdentity
}
from
'../src/integrations/meyer/identity-retry'
;
import
{
mkdtemp
,
writeFile
,
rm
}
from
'node:fs/promises'
;
import
{
join
}
from
'node:path'
;
import
{
tmpdir
}
from
'node:os'
;
...
...
@@ -69,6 +70,57 @@ 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
(
'identity recovery CLI verifies live binding and full PAC identity, then resumes the normal tick'
,
async
()
=>
{
await
hostRecord
();
const
directory
=
await
mkdtemp
(
join
(
tmpdir
(),
'meyer-identity-pg-'
));
const
state
=
new
MeyerStore
(
join
(
directory
,
'state'
));
let
image
=
{
...
meyerImage
(),
patientId
:
null
as
string
|
null
};
const
server
=
createServer
(
async
(
req
,
res
)
=>
{
for
await
(
const
_chunk
of
req
)
{
/* 不记录凭据或请求原文。 */
}
res
.
setHeader
(
'content-type'
,
'application/json'
);
const
data
=
req
.
url
?.
includes
(
'/tokens/pub'
)
?
{
accessToken
:
'local-token'
,
accessExpiresAt
:
new
Date
(
Date
.
now
()
+
3600000
).
toISOString
()
}
:
req
.
url
?.
endsWith
(
'/query-by-clinic'
)
?
{
total
:
1
,
page
:
1
,
size
:
100
,
images
:
[
image
]
}
:
meyerSymptom
();
res
.
end
(
JSON
.
stringify
({
code
:
200
,
data
}));
});
const
previousKey
=
process
.
env
.
MEYER_APP_KEY
,
previousSecret
=
process
.
env
.
MEYER_APP_SECRET
;
const
output
=
jest
.
spyOn
(
console
,
'log'
).
mockImplementation
(()
=>
{});
try
{
await
new
Promise
<
void
>
(
r
=>
server
.
listen
(
0
,
'127.0.0.1'
,
r
));
const
port
=
(
server
.
address
()
as
{
port
:
number
}).
port
;
const
{
scopes
,
...
fields
}
=
meyerConfig
();
await
writeFile
(
join
(
directory
,
'config.json'
),
JSON
.
stringify
({
...
fields
,
timeStart
:
'2026-10-10 09:00:00'
,
baseUrl
:
`http://127.0.0.1:
${
port
}
/api/v1`
,
clinicIds
:
scopes
.
map
(
s
=>
s
.
clinicId
)
}));
process
.
env
.
MEYER_APP_KEY
=
'local'
;
process
.
env
.
MEYER_APP_SECRET
=
'local'
;
const
key
=
imageKey
(
scopes
[
0
]
!
,
image
.
imageId
),
pending
=
new
MeyerStore
(
join
(
state
.
directory
,
'pending'
));
const
p
=
{
scope
:
scopes
[
0
]
!
,
image
,
attempts
:
9
,
identityAttempts
:
8
,
identityIssue
:
'missing patientId'
,
nextRetryAt
:
new
Date
().
toISOString
()
};
await
pending
.
write
(
key
,
p
);
await
retireIdentity
(
state
,
key
,
p
,
Date
.
now
());
const
args
=
[
`--config=
${
join
(
directory
,
'config.json'
)}
`
,
`--state-dir=
${
state
.
directory
}
`
];
const
recover
=
[...
args
,
`--resume-identity=
${
image
.
imageId
}
`
];
await
expect
(
runMeyerCli
(
recover
)).
rejects
.
toThrow
(
'not ready'
);
image
=
{
...
image
,
patientId
:
'patient-2'
};
await
prisma
.
patient
.
create
({
data
:
{
hostId
,
tenantId
:
'friday-market'
,
sourceUnit
:
'brand-b'
,
externalId
:
'patient-2'
}
});
await
expect
(
runMeyerCli
([...
recover
,
'--apply'
])).
rejects
.
toThrow
(
'patient not present'
);
expect
(
await
pending
.
keys
()).
toEqual
([]);
image
=
{
...
image
,
patientId
:
'patient-1'
};
expect
(
await
runMeyerCli
(
recover
)).
toBe
(
0
);
expect
(
await
identityReviewStore
(
state
).
read
(
key
)).
not
.
toBeNull
();
expect
(
await
runMeyerCli
([...
recover
,
'--apply'
])).
toBe
(
0
);
expect
(
await
pending
.
read
(
key
)).
toMatchObject
({
identityAttempts
:
0
});
expect
(
await
prisma
.
patientFact
.
count
({
where
:
{
hostId
}
})).
toBe
(
0
);
expect
(
await
runMeyerCli
([...
args
,
'--tick'
,
'--apply'
])).
toBe
(
0
);
expect
(
await
pending
.
keys
()).
toEqual
([]);
expect
(
await
prisma
.
patientFact
.
count
({
where
:
{
hostId
,
patientId
,
type
:
'image_record'
}
})).
toBeGreaterThan
(
0
);
expect
(
await
prisma
.
patientFact
.
count
({
where
:
{
hostId
,
patientId
,
type
:
'diagnosis_record'
}
})).
toBe
(
2
);
expect
(
await
prisma
.
patientFact
.
count
({
where
:
{
hostId
,
patientId
:
otherPatientId
}
})).
toBe
(
0
);
}
finally
{
output
.
mockRestore
();
if
(
previousKey
===
undefined
)
delete
process
.
env
.
MEYER_APP_KEY
;
else
process
.
env
.
MEYER_APP_KEY
=
previousKey
;
if
(
previousSecret
===
undefined
)
delete
process
.
env
.
MEYER_APP_SECRET
;
else
process
.
env
.
MEYER_APP_SECRET
=
previousSecret
;
server
.
closeAllConnections
();
await
new
Promise
<
void
>
(
r
=>
server
.
close
(()
=>
r
()));
await
rm
(
directory
,
{
recursive
:
true
,
force
:
true
});
}
});
test
(
'automatic clinic mapping isolates missing and conflicting clinics without losing valid scopes'
,
async
()
=>
{
await
hostRecord
();
const
result
=
await
loadMeyerScopeResults
([
'missing-clinic'
,
'clinic-a'
]);
...
...
Write
Preview
Markdown
is supported
0%
Try again
or
attach a new file
Attach a file
Cancel
You are about to add
0
people
to the discussion. Proceed with caution.
Finish editing this message first!
Cancel
Please
register
or
sign in
to comment