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
7d1e6bad
Commit
7d1e6bad
authored
Oct 21, 2026
by
luoqi
Browse files
Options
Browse Files
Download
Plain Diff
Merge bounded Meyer identity retries into test
parents
c888729e
ff1c0399
Pipeline
#3848
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 @
7d1e6bad
...
...
@@ -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 @
7d1e6bad
...
...
@@ -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 @
7d1e6bad
...
...
@@ -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 @
7d1e6bad
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 @
7d1e6bad
...
...
@@ -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 @
7d1e6bad
...
...
@@ -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 @
7d1e6bad
...
...
@@ -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 @
7d1e6bad
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 @
7d1e6bad
...
...
@@ -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