diff --git a/docs/03-目标架构设计.md b/docs/03-目标架构设计.md index 6e4f3c4..4865a99 100644 --- a/docs/03-目标架构设计.md +++ b/docs/03-目标架构设计.md @@ -159,6 +159,12 @@ type ManageInfo struct { DSFMC, OperateUnit, OperateUser string } ### 4.6 `pipeline` —— 投递编排 +T-213 已落地可复用的 `HealthRecordUpsertService`。它接收 `source.HealthRecordTask`，通过注入的 `Converter`、`HealthRecordClient`、`IdempotencyStore`、`ReportSink` 和 `PHISStatusWriter` 完成单条健康档案编排；HTTP handler 与未来 worker 必须复用该服务，不复制分支规则。 + +健康档案采用 query-first：映射校验通过后，以稳定 `checkId + idCard` 的哈希键原子获取幂等租约，再按身份证调用 JKDA00002。明确 0 条才调用 JKDA00001；恰好 1 条且身份证、状态、机构、责任医生、`phrId` 均安全时调用 JKDA00003；多条或任一授权条件不满足均转 `manual_review`。查询失败、创建失败、更新失败均不切换另一种写操作。 + +当前幂等接口是带 owner token 的 `Acquire/Complete/Release` 契约，完成和释放必须匹配租约 token，防止过期 worker 改写新租约；T-213 只用假实现验证占用和完成语义，Redis/文件持久化、租约超时与恢复属于阶段 3。报告与 PHIS 状态同样仅定义端口，真实回写由 T-214 实现。CHIS 已明确写入成功后，必须先完成幂等记录再发布 `done`；完成失败时保留业务结果 `done`、返回基础设施错误且不发送成功通知，调用方不得据此重放 CHIS 写入。副作用失败通过 `NotificationError` 分别暴露 `PriorErr/ReportErr/StatusErr`，并只携带失败端的补偿载荷，补偿时不重新执行 upsert。 + 单条投递流程（继承旧项目 worker 经验）： ``` @@ -171,13 +177,13 @@ type ManageInfo struct { DSFMC, OperateUnit, OperateUser string } ``` - **重试分类**（移植并适配 OSI）：网络错误 / HTTP 5xx / 429 / `code==405`(超时) 可重试；参数错、权限错、映射校验错不可重试；退避 `2s*attempt`，最多 3 次。 -- **幂等**：键 = `checkId`（替代旧项目报文哈希）；存储 redis 优先、文件降级。 +- **幂等**：健康档案键 = `sha256(checkId + "|" + idCard)`，原子获取带 owner token 的租约；存储 redis 优先、文件降级。 - **熔断**：连续网络失败达阈值则暂停（阈值/休眠秒可配）。 - **批次报告**：`total/success/failed/skipped/retry`、失败 Top、逐条明细（沿用旧项目格式）。 ### 4.7 `observ` / `store` —— 可观测与存储 -- 一套 report log（收敛旧项目的 apitrace/reportlog/snapshot 三套）：记录每次 PHIS 输入、映射结果、OSI 请求/响应、最终判定。 +- 一套 report log（收敛旧项目的 apitrace/reportlog/snapshot 三套）：普通日志只记录脱敏标识、serviceId、响应码和最终判定；原始 PHIS 输入与完整 OSI 请求/响应不得写入普通日志。 - Redis 优先、本地 JSONL 降级；Redis 启动失败不阻断（沿用旧项目）。 --- diff --git a/docs/current-state.md b/docs/current-state.md index 7a0014b..81f55a8 100644 --- a/docs/current-state.md +++ b/docs/current-state.md @@ -6,14 +6,14 @@ ## 当前快照 - 日期：2026-07-16 -- 阶段：**T-212 PHIS 健康档案转换器已完成；下一步 T-213 upsert 应用编排，真实写入验收由 T-215 单独受阻**；T-303 老年人生活自理能力评估查询代码完成、真实联调受阻 +- 阶段：**T-212 转换器与 T-213 健康档案 upsert 应用编排已完成；下一步 T-204 暴露 HTTP upsert，真实写入验收由 T-215 单独受阻**；T-303 老年人生活自理能力评估查询代码完成、真实联调受阻 - 技术栈：Go 1.24 单二进制；`main.go -mode server|deliver`；配置读取使用 viper，支持环境变量覆盖；OSI 客户端已具备签名、信封、传输、基础判码、JKDA00002 Find、JKDA00005 FindRqbj（人群分类，已按实测 auto 路径校准）、JKDA00001 Create、JKDA00003 Update、LNRZLPG00002 老年人生活自理能力查询（待真实联调），以及 WGDZ/ZRYS/YPML/CXJG 四个公开查询薄封装 -- 生产代码：新增 `source/health_record.go` 承接 PHIS 档案 DTO 并剥离医生凭据；`mapping/health_record.go` 已能把 PHIS 主体、既往史和生活环境转换为完整 JKDA 写入请求，校验稳定 archId、格式/码表及医生/机构主数据；OSI 写入保留逐档案 `operateUser`，配置继续控制 DSFMC/operateUnit；`pipeline/` 等业务模块仍待 T-213 建立 +- 生产代码：`source/health_record.go` 承接 PHIS 档案 DTO 并剥离医生凭据；`mapping/health_record.go` 把 PHIS 主体、既往史和生活环境转换为 JKDA 写入请求；`pipeline/health_record_upsert.go` 已实现可注入、可被 HTTP/worker 复用的 query-first upsert，覆盖原子幂等占用、创建/更新/人工处理/失败分类和脱敏事件 - 联调现实：**JKDA00002 个人档案查询已用 Go 侧真实请求打通**，返回 `code="01" message="操作成功" data_count=1`；**公开查询 WGDZ00001/ZRYS00001/CXJG00002 已用真实档案主数据验证通过**，均返回 `code="01"` 且数组非空；药品目录 YPML00001 已完成客户端封装和单测，尚未做真实药品关键字样本验证；**JKDA00001 create 未跑真实请求**，避免在没有安全测试居民/写入授权时污染平台档案 - 测试：`go test ./...` 通过；当前测试覆盖 mode 解析、配置加载与环境变量覆盖、MD5 签名、请求头组装、JSON POST 传输、头名大小写保留、identity 响应编码声明、超时配置、SOCKS5 代理地址校验、信封结构、serviceId 路由、成功/重试判码、Client.Call 请求与响应解析、JKDA00002 查询响应契约、JKDA00001/00003 创建更新请求契约与客户端方法、JKDA00002 Find、JKDA00005 FindRqbj（personSign/idCard/phrId）、公开查询四接口、公开查询 HTTP API、字典缓存快照与持久化失败不阻断、映射码表双向查找、民族 01~56 完整性、健康档案映射必填/码表校验、主数据名称反查、创建请求组装、checkId 确定性、docx/联调风格映射样本基线、JKDA00002 验证入口 - 标准启动路径：`./init.sh` 已配置三步：依赖下载、`go test ./...`、`go run . -mode server -config config.yaml.example` - 标准验证路径：`go test ./...`、`go build ./...` -- 当前 blocker：T-215 真实 create/update 验收缺少可写入沙箱的安全测试居民资料或明确写入授权；该 blocker 不影响先做 T-212 转换器和 T-213 假依赖编排。PHIS `archId/businessId/phrId/empiId` 的稳定主键语义、逐档案 `operateUser` 规则也需在 T-212/T-215 中确认。软限制：当前机器从 Git Bash 启动 Go 会出现标准库路径/构建缓存权限异常，`init.sh` 无法完整跑完；PowerShell 下等价 Go 命令和真实查询通过。厂家侧 B1/B2/B4/B5 契约缺口只影响阶段 4 +- 当前 blocker：T-215 真实 create/update 验收缺少可写入沙箱的安全测试居民资料或明确写入授权；PHIS 源键已在 T-212 固化为 `archId`，`addressNumber`、`createDate` 和逐档案 `operateUser` 仍需 T-215 写入后回查确认。软限制：当前机器从 Git Bash 启动 Go 会出现标准库路径/构建缓存权限异常，`init.sh` 无法完整跑完；PowerShell 下等价 Go 命令和真实查询通过。厂家侧 B1/B2/B4/B5 契约缺口只影响阶段 4 ## 当前目录要点 @@ -45,7 +45,7 @@ | `handler/elderly.go` | 已有 | `GET /api/elderly/self-care` 老年人生活自理能力评估查询（待真实联调） | | `handler/public.go` | 已有 | 公开查询端点 `GET /api/dictionaries/grid-addresses`、`/doctors`、`/drugs`、`/orgs` | | `server.go` `osi_client.go` | 已有 | server 模式起 HTTP 服务、注册路由；`buildOSIClient` 共用构造（verify 与 server 复用） | -| `pipeline/` 等 | 待建 | 后续任务 | +| `pipeline/health_record_upsert.go` | 已有 | T-213 query-first upsert 应用服务；外部客户端、原子幂等、报告与 PHIS 状态均接口注入，持久化待阶段 3 | ## 已验证事实（写代码时直接依赖） diff --git a/pipeline/health_record_upsert.go b/pipeline/health_record_upsert.go new file mode 100644 index 0000000..84fc60a --- /dev/null +++ b/pipeline/health_record_upsert.go @@ -0,0 +1,312 @@ +package pipeline + +import ( + "context" + "crypto/sha256" + "errors" + "fmt" + "strings" + + "chis_osi/contract" + "chis_osi/mapping" + "chis_osi/osi" + "chis_osi/source" +) + +type Status string + +const ( + StatusDone Status = "done" + StatusRetry Status = "retry" + StatusFailed Status = "failed" + StatusManualReview Status = "manual_review" +) + +type Action string + +const ( + ActionNone Action = "none" + ActionCreate Action = "create" + ActionUpdate Action = "update" + ActionSkip Action = "skip" +) + +type Outcome struct { + Status Status + Action Action + Reason string + ServiceID string + ResponseCode string + PHRIDHint string +} + +type UpsertEvent struct { + TraceID string + Status Status + Action Action + Reason string + ServiceID string + ResponseCode string + PHRIDHint string +} + +type PHISStatusUpdate struct { + SourceRecordID string + BusinessID string + Status Status + Reason string +} + +type UpdateTarget struct { + PHRID string + EMPIID string + CheckID string + IDCard string + ManaUnitID string + ManaDoctorID string + Status string +} + +type Converter interface { + ConvertHealthRecord(source.HealthRecordTask) (contract.HealthRecordCreate, []mapping.ValidationError) +} + +type ConverterFunc func(source.HealthRecordTask) (contract.HealthRecordCreate, []mapping.ValidationError) + +func (f ConverterFunc) ConvertHealthRecord(task source.HealthRecordTask) (contract.HealthRecordCreate, []mapping.ValidationError) { + return f(task) +} + +type HealthRecordClient interface { + FindHealthRecord(context.Context, osi.FindHealthRecordQuery) ([]contract.FindHealthRecord, osi.Result, error) + CreateHealthRecord(context.Context, contract.HealthRecordCreate) (contract.HealthRecordSaveResult, osi.Result, error) + UpdateHealthRecord(context.Context, contract.HealthRecordCreate) (contract.HealthRecordSaveResult, osi.Result, error) +} + +type IdempotencyStore interface { + Acquire(context.Context, string) (IdempotencyLease, error) + Complete(context.Context, string, string) error + Release(context.Context, string, string) error +} + +type IdempotencyState string + +const ( + IdempotencyAcquired IdempotencyState = "acquired" + IdempotencyCompleted IdempotencyState = "completed" + IdempotencyInProgress IdempotencyState = "in_progress" +) + +type IdempotencyLease struct { + State IdempotencyState + Token string +} + +type NotificationError struct { + Event *UpsertEvent + Update *PHISStatusUpdate + PriorErr error + ReportErr error + StatusErr error +} + +func (e *NotificationError) Error() string { + return "finish upsert side effects: " + e.Unwrap().Error() +} +func (e *NotificationError) Unwrap() error { return errors.Join(e.PriorErr, e.ReportErr, e.StatusErr) } + +type ReportSink interface { + Publish(context.Context, UpsertEvent) error +} + +type PHISStatusWriter interface { + WriteStatus(context.Context, PHISStatusUpdate) error +} + +type Dependencies struct { + Converter Converter + Client HealthRecordClient + Idempotency IdempotencyStore + Reports ReportSink + PHISStatuses PHISStatusWriter +} + +type HealthRecordUpsertService struct { + deps Dependencies +} + +func NewHealthRecordUpsertService(deps Dependencies) *HealthRecordUpsertService { + return &HealthRecordUpsertService{deps: deps} +} + +func (s *HealthRecordUpsertService) Upsert(ctx context.Context, task source.HealthRecordTask) (Outcome, error) { + if err := s.validateDependencies(); err != nil { + return Outcome{}, err + } + + req, validationErrors := s.deps.Converter.ConvertHealthRecord(task) + if len(validationErrors) > 0 { + return s.finish(ctx, task, Outcome{Status: StatusFailed, Action: ActionNone, Reason: "mapping validation failed: " + validationErrors[0].Field}) + } + + idempotencyKey := healthRecordIdempotencyKey(req.HealthRecord.CheckID, req.HealthRecord.IDCard) + lease, err := s.deps.Idempotency.Acquire(ctx, idempotencyKey) + if err != nil { + return s.finish(ctx, task, Outcome{Status: StatusRetry, Action: ActionNone, Reason: "idempotency lookup failed"}, err) + } + if lease.State == IdempotencyCompleted { + return s.finish(ctx, task, Outcome{Status: StatusDone, Action: ActionSkip, Reason: "already completed"}) + } + if lease.State == IdempotencyInProgress { + return s.finish(ctx, task, Outcome{Status: StatusRetry, Action: ActionSkip, Reason: "another delivery is in progress"}) + } + if lease.State != IdempotencyAcquired || strings.TrimSpace(lease.Token) == "" { + return s.finish(ctx, task, Outcome{Status: StatusRetry, Action: ActionNone, Reason: "invalid idempotency lease"}) + } + + records, queryResult, err := s.deps.Client.FindHealthRecord(ctx, osi.FindHealthRecordQuery{IDCard: req.HealthRecord.IDCard}) + if err != nil || !queryResult.Success { + status := classifyFailure(queryResult) + return s.finishAcquired(ctx, task, idempotencyKey, lease.Token, Outcome{Status: status, Action: ActionNone, Reason: "CHIS query failed", ServiceID: osi.ServiceIDJKDAFind, ResponseCode: queryResult.Code}) + } + if records == nil { + return s.finishAcquired(ctx, task, idempotencyKey, lease.Token, Outcome{Status: StatusFailed, Action: ActionNone, Reason: "CHIS query returned null or missing data", ServiceID: osi.ServiceIDJKDAFind, ResponseCode: queryResult.Code}) + } + + if len(records) > 1 { + return s.finishAcquired(ctx, task, idempotencyKey, lease.Token, Outcome{Status: StatusManualReview, Action: ActionNone, Reason: "multiple CHIS health records found", ServiceID: osi.ServiceIDJKDAFind, ResponseCode: queryResult.Code}) + } + if len(records) == 0 { + return s.create(ctx, task, idempotencyKey, lease.Token, req) + } + + target := newUpdateTarget(records[0].HealthRecord) + if reason := validateUpdateTarget(target, req); reason != "" { + return s.finishAcquired(ctx, task, idempotencyKey, lease.Token, Outcome{Status: StatusManualReview, Action: ActionNone, Reason: reason, ServiceID: osi.ServiceIDJKDAFind, ResponseCode: queryResult.Code, PHRIDHint: maskIdentifier(target.PHRID)}) + } + req.BaseInfo.PHRID = target.PHRID + req.HealthRecord.PhrID = target.PHRID + return s.update(ctx, task, idempotencyKey, lease.Token, req) +} + +func (s *HealthRecordUpsertService) create(ctx context.Context, task source.HealthRecordTask, key, token string, req contract.HealthRecordCreate) (Outcome, error) { + saved, result, err := s.deps.Client.CreateHealthRecord(ctx, req) + outcome := Outcome{Status: StatusDone, Action: ActionCreate, ServiceID: osi.ServiceIDJKDACreate, ResponseCode: result.Code, PHRIDHint: maskIdentifier(saved.PhrID)} + if err != nil || !result.Success { + outcome.Status = classifyFailure(result) + outcome.Reason = "CHIS create failed" + return s.finishAcquired(ctx, task, key, token, outcome) + } + return s.complete(ctx, task, key, token, outcome) +} + +func (s *HealthRecordUpsertService) update(ctx context.Context, task source.HealthRecordTask, key, token string, req contract.HealthRecordCreate) (Outcome, error) { + saved, result, err := s.deps.Client.UpdateHealthRecord(ctx, req) + phrID := saved.PhrID + if phrID == "" { + phrID = req.HealthRecord.PhrID + } + outcome := Outcome{Status: StatusDone, Action: ActionUpdate, ServiceID: osi.ServiceIDJKDAUpdate, ResponseCode: result.Code, PHRIDHint: maskIdentifier(phrID)} + if err != nil || !result.Success { + outcome.Status = classifyFailure(result) + outcome.Reason = "CHIS update failed" + return s.finishAcquired(ctx, task, key, token, outcome) + } + return s.complete(ctx, task, key, token, outcome) +} + +func (s *HealthRecordUpsertService) complete(ctx context.Context, task source.HealthRecordTask, key, token string, outcome Outcome) (Outcome, error) { + if err := s.deps.Idempotency.Complete(ctx, key, token); err != nil { + return outcome, fmt.Errorf("persist successful CHIS write before notification: %w", err) + } + return s.finish(ctx, task, outcome) +} + +func (s *HealthRecordUpsertService) finishAcquired(ctx context.Context, task source.HealthRecordTask, key, token string, outcome Outcome) (Outcome, error) { + return s.finish(ctx, task, outcome, s.deps.Idempotency.Release(ctx, key, token)) +} + +func (s *HealthRecordUpsertService) finish(ctx context.Context, task source.HealthRecordTask, outcome Outcome, prior ...error) (Outcome, error) { + event := UpsertEvent{ + TraceID: traceID(task.SourceRecordID(), task.BusinessID), Status: outcome.Status, + Action: outcome.Action, Reason: outcome.Reason, ServiceID: outcome.ServiceID, + ResponseCode: outcome.ResponseCode, PHRIDHint: outcome.PHRIDHint, + } + status := PHISStatusUpdate{SourceRecordID: task.SourceRecordID(), BusinessID: task.BusinessID, Status: outcome.Status, Reason: outcome.Reason} + priorErr := errors.Join(prior...) + reportErr := s.deps.Reports.Publish(ctx, event) + statusErr := s.deps.PHISStatuses.WriteStatus(ctx, status) + if priorErr != nil || reportErr != nil || statusErr != nil { + notificationErr := &NotificationError{PriorErr: priorErr, ReportErr: reportErr, StatusErr: statusErr} + if reportErr != nil { + notificationErr.Event = &event + } + if statusErr != nil { + notificationErr.Update = &status + } + return outcome, notificationErr + } + return outcome, nil +} + +func (s *HealthRecordUpsertService) validateDependencies() error { + if s == nil || s.deps.Converter == nil || s.deps.Client == nil || s.deps.Idempotency == nil || s.deps.Reports == nil || s.deps.PHISStatuses == nil { + return fmt.Errorf("health record upsert dependencies are incomplete") + } + return nil +} + +func classifyFailure(result osi.Result) Status { + if result.Retryable { + return StatusRetry + } + return StatusFailed +} + +func newUpdateTarget(record contract.HealthRecord) UpdateTarget { + checkID := "" + if record.CheckID != nil { + checkID = *record.CheckID + } + return UpdateTarget{ + PHRID: record.PhrID, EMPIID: record.EmpiID, CheckID: checkID, IDCard: record.IDCard, + ManaUnitID: record.ManaUnitID, ManaDoctorID: record.ManaDoctorID, Status: record.Status, + } +} + +func validateUpdateTarget(target UpdateTarget, req contract.HealthRecordCreate) string { + if strings.TrimSpace(target.IDCard) != strings.TrimSpace(req.HealthRecord.IDCard) { + return "CHIS identity does not exactly match source" + } + if strings.TrimSpace(target.PHRID) == "" { + return "CHIS record is missing phrId" + } + if strings.TrimSpace(target.Status) != "1" { + return "CHIS record status is not updateable" + } + if strings.TrimSpace(target.ManaUnitID) != strings.TrimSpace(req.HealthRecord.ManaUnitID) { + return "CHIS record belongs to another organization" + } + if strings.TrimSpace(target.ManaDoctorID) != strings.TrimSpace(req.HealthRecord.ManaDoctorID) { + return "CHIS record belongs to another doctor" + } + return "" +} + +func healthRecordIdempotencyKey(checkID, idCard string) string { + sum := sha256.Sum256([]byte(strings.TrimSpace(checkID) + "|" + strings.TrimSpace(idCard))) + return fmt.Sprintf("health-record:%x", sum[:]) +} + +func traceID(sourceRecordID, businessID string) string { + sum := sha256.Sum256([]byte(strings.TrimSpace(sourceRecordID) + "|" + strings.TrimSpace(businessID))) + return fmt.Sprintf("hr-%x", sum[:8]) +} + +func maskIdentifier(value string) string { + value = strings.TrimSpace(value) + if len(value) <= 4 { + return "****" + } + return "****" + value[len(value)-4:] +} diff --git a/pipeline/health_record_upsert_test.go b/pipeline/health_record_upsert_test.go new file mode 100644 index 0000000..da3dd69 --- /dev/null +++ b/pipeline/health_record_upsert_test.go @@ -0,0 +1,302 @@ +package pipeline + +import ( + "context" + "errors" + "testing" + + "chis_osi/contract" + "chis_osi/mapping" + "chis_osi/osi" + "chis_osi/source" +) + +func TestHealthRecordUpsertCreatesWhenQueryIsEmpty(t *testing.T) { + client := &fakeHealthRecordClient{rows: []contract.FindHealthRecord{}, findResult: osi.Result{Success: true}, saveResult: osi.Result{Success: true, Code: "01"}, saved: contract.HealthRecordSaveResult{PhrID: "PHR-NEW"}} + store := &fakeIdempotencyStore{} + reports := &fakeReportSink{} + statuses := &fakeStatusWriter{} + service := newTestService(client, store, reports, statuses) + + outcome, err := service.Upsert(context.Background(), testTask()) + if err != nil || outcome.Status != StatusDone || outcome.Action != ActionCreate { + t.Fatalf("outcome=%#v err=%v", outcome, err) + } + if client.findCalls != 1 || client.createCalls != 1 || client.updateCalls != 0 || !store.marked { + t.Fatalf("calls find=%d create=%d update=%d marked=%v", client.findCalls, client.createCalls, client.updateCalls, store.marked) + } + assertNotifications(t, reports, statuses, StatusDone, ActionCreate) +} + +func TestHealthRecordUpsertUpdatesOneMatchingRecord(t *testing.T) { + client := &fakeHealthRecordClient{ + findResult: osi.Result{Success: true}, + rows: []contract.FindHealthRecord{{HealthRecord: matchingTarget()}}, + saveResult: osi.Result{Success: true, Code: "01"}, + saved: contract.HealthRecordSaveResult{PhrID: "PHR-EXISTING"}, + } + service := newTestService(client, &fakeIdempotencyStore{}, &fakeReportSink{}, &fakeStatusWriter{}) + + outcome, err := service.Upsert(context.Background(), testTask()) + if err != nil || outcome.Status != StatusDone || outcome.Action != ActionUpdate { + t.Fatalf("outcome=%#v err=%v", outcome, err) + } + if client.updateCalls != 1 || client.createCalls != 0 { + t.Fatalf("create=%d update=%d", client.createCalls, client.updateCalls) + } + if client.updateReq.BaseInfo.PHRID != "PHR-EXISTING" || client.updateReq.HealthRecord.PhrID != "PHR-EXISTING" { + t.Fatalf("update phrId not merged: %#v", client.updateReq) + } + if client.updateReq.HealthRecord.CheckID != "CHECK-STABLE" { + t.Fatalf("source checkId was overwritten: %q", client.updateReq.HealthRecord.CheckID) + } +} + +func TestHealthRecordUpsertRoutesUnsafeMatchesToManualReview(t *testing.T) { + tests := []struct { + name string + rows []contract.FindHealthRecord + }{ + {name: "multiple", rows: []contract.FindHealthRecord{{HealthRecord: matchingTarget()}, {HealthRecord: matchingTarget()}}}, + {name: "cross unit", rows: []contract.FindHealthRecord{{HealthRecord: targetWith(func(r *contract.HealthRecord) { r.ManaUnitID = "UNIT-OTHER" })}}}, + {name: "cross doctor", rows: []contract.FindHealthRecord{{HealthRecord: targetWith(func(r *contract.HealthRecord) { r.ManaDoctorID = "DOC-OTHER" })}}}, + {name: "inactive", rows: []contract.FindHealthRecord{{HealthRecord: targetWith(func(r *contract.HealthRecord) { r.Status = "0" })}}}, + {name: "id mismatch", rows: []contract.FindHealthRecord{{HealthRecord: targetWith(func(r *contract.HealthRecord) { r.IDCard = "440000********9999" })}}}, + {name: "missing phrId", rows: []contract.FindHealthRecord{{HealthRecord: targetWith(func(r *contract.HealthRecord) { r.PhrID = "" })}}}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + client := &fakeHealthRecordClient{rows: tt.rows, findResult: osi.Result{Success: true}} + service := newTestService(client, &fakeIdempotencyStore{}, &fakeReportSink{}, &fakeStatusWriter{}) + outcome, err := service.Upsert(context.Background(), testTask()) + if err != nil || outcome.Status != StatusManualReview || client.createCalls != 0 || client.updateCalls != 0 { + t.Fatalf("outcome=%#v err=%v create=%d update=%d", outcome, err, client.createCalls, client.updateCalls) + } + }) + } +} + +func TestHealthRecordUpsertNeverCreatesAfterQueryFailure(t *testing.T) { + for _, retryable := range []bool{false, true} { + client := &fakeHealthRecordClient{findResult: osi.Result{Retryable: retryable, Code: "405"}, findErr: errors.New("query failed")} + store := &fakeIdempotencyStore{} + service := newTestService(client, store, &fakeReportSink{}, &fakeStatusWriter{}) + outcome, err := service.Upsert(context.Background(), testTask()) + want := StatusFailed + if retryable { + want = StatusRetry + } + if err != nil || outcome.Status != want || client.createCalls != 0 || client.updateCalls != 0 || !store.released { + t.Fatalf("retryable=%v outcome=%#v err=%v", retryable, outcome, err) + } + } +} + +func TestHealthRecordUpsertNeverCreatesForNullQueryData(t *testing.T) { + client := &fakeHealthRecordClient{rows: nil, findResult: osi.Result{Success: true, Code: "01"}} + service := newTestService(client, &fakeIdempotencyStore{}, &fakeReportSink{}, &fakeStatusWriter{}) + outcome, err := service.Upsert(context.Background(), testTask()) + if err != nil || outcome.Status != StatusFailed || client.createCalls != 0 { + t.Fatalf("outcome=%#v err=%v create=%d", outcome, err, client.createCalls) + } +} + +func TestHealthRecordUpsertClassifiesWriteFailureWithoutFallback(t *testing.T) { + tests := []struct { + name string + rows []contract.FindHealthRecord + retryable bool + action Action + }{ + {name: "create retry", rows: []contract.FindHealthRecord{}, retryable: true, action: ActionCreate}, + {name: "create fail", rows: []contract.FindHealthRecord{}, action: ActionCreate}, + {name: "update retry", rows: []contract.FindHealthRecord{{HealthRecord: matchingTarget()}}, retryable: true, action: ActionUpdate}, + {name: "update fail", rows: []contract.FindHealthRecord{{HealthRecord: matchingTarget()}}, action: ActionUpdate}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + client := &fakeHealthRecordClient{rows: tt.rows, findResult: osi.Result{Success: true}, saveResult: osi.Result{Retryable: tt.retryable}, saveErr: errors.New("write failed")} + service := newTestService(client, &fakeIdempotencyStore{}, &fakeReportSink{}, &fakeStatusWriter{}) + outcome, err := service.Upsert(context.Background(), testTask()) + want := StatusFailed + if tt.retryable { + want = StatusRetry + } + if err != nil || outcome.Status != want || outcome.Action != tt.action { + t.Fatalf("outcome=%#v err=%v", outcome, err) + } + if client.createCalls+client.updateCalls != 1 { + t.Fatalf("unexpected write fallback: create=%d update=%d", client.createCalls, client.updateCalls) + } + }) + } +} + +func TestHealthRecordUpsertSkipsCompletedIdempotencyKey(t *testing.T) { + client := &fakeHealthRecordClient{} + service := newTestService(client, &fakeIdempotencyStore{completed: true}, &fakeReportSink{}, &fakeStatusWriter{}) + outcome, err := service.Upsert(context.Background(), testTask()) + if err != nil || outcome.Status != StatusDone || outcome.Action != ActionSkip || client.findCalls != 0 { + t.Fatalf("outcome=%#v err=%v find=%d", outcome, err, client.findCalls) + } +} + +func TestHealthRecordUpsertDoesNotRaceAnInProgressDelivery(t *testing.T) { + client := &fakeHealthRecordClient{} + service := newTestService(client, &fakeIdempotencyStore{inProgress: true}, &fakeReportSink{}, &fakeStatusWriter{}) + outcome, err := service.Upsert(context.Background(), testTask()) + if err != nil || outcome.Status != StatusRetry || outcome.Action != ActionSkip || client.findCalls != 0 { + t.Fatalf("outcome=%#v err=%v find=%d", outcome, err, client.findCalls) + } +} + +func TestHealthRecordUpsertRejectsMappingErrorsBeforeOSI(t *testing.T) { + client := &fakeHealthRecordClient{} + service := NewHealthRecordUpsertService(Dependencies{ + Converter: ConverterFunc(func(source.HealthRecordTask) (contract.HealthRecordCreate, []mapping.ValidationError) { + return contract.HealthRecordCreate{}, []mapping.ValidationError{{Field: "idCard", Reason: "required"}} + }), + Client: client, Idempotency: &fakeIdempotencyStore{}, Reports: &fakeReportSink{}, PHISStatuses: &fakeStatusWriter{}, + }) + outcome, err := service.Upsert(context.Background(), testTask()) + if err != nil || outcome.Status != StatusFailed || client.findCalls != 0 { + t.Fatalf("outcome=%#v err=%v find=%d", outcome, err, client.findCalls) + } +} + +func TestHealthRecordUpsertDoesNotNotifyDoneWhenCompletionFails(t *testing.T) { + client := &fakeHealthRecordClient{rows: []contract.FindHealthRecord{}, findResult: osi.Result{Success: true}, saveResult: osi.Result{Success: true}} + store := &fakeIdempotencyStore{completeErr: errors.New("store unavailable")} + reports := &fakeReportSink{} + statuses := &fakeStatusWriter{} + service := newTestService(client, store, reports, statuses) + outcome, err := service.Upsert(context.Background(), testTask()) + if err == nil || outcome.Status != StatusDone || len(reports.events) != 0 || len(statuses.updates) != 0 { + t.Fatalf("outcome=%#v err=%v reports=%#v statuses=%#v", outcome, err, reports.events, statuses.updates) + } +} + +func TestHealthRecordUpsertNotificationErrorOnlyCarriesFailedSink(t *testing.T) { + client := &fakeHealthRecordClient{rows: []contract.FindHealthRecord{}, findResult: osi.Result{Success: true}, saveResult: osi.Result{Success: true}} + statuses := &fakeStatusWriter{err: errors.New("PHIS unavailable")} + service := newTestService(client, &fakeIdempotencyStore{}, &fakeReportSink{}, statuses) + + outcome, err := service.Upsert(context.Background(), testTask()) + var notificationErr *NotificationError + if outcome.Status != StatusDone || !errors.As(err, ¬ificationErr) { + t.Fatalf("outcome=%#v err=%v", outcome, err) + } + if notificationErr.Event != nil || notificationErr.Update == nil || notificationErr.ReportErr != nil || notificationErr.StatusErr == nil { + t.Fatalf("notification error = %#v", notificationErr) + } +} + +func newTestService(client *fakeHealthRecordClient, store *fakeIdempotencyStore, reports *fakeReportSink, statuses *fakeStatusWriter) *HealthRecordUpsertService { + return NewHealthRecordUpsertService(Dependencies{ + Converter: ConverterFunc(func(source.HealthRecordTask) (contract.HealthRecordCreate, []mapping.ValidationError) { + return testRequest(), nil + }), + Client: client, Idempotency: store, Reports: reports, PHISStatuses: statuses, + }) +} + +func testTask() source.HealthRecordTask { + return source.HealthRecordTask{ArchID: "ARCH-1", BusinessID: "BUS-1"} +} + +func testRequest() contract.HealthRecordCreate { + return contract.HealthRecordCreate{ + BaseInfo: contract.HealthRecordBaseInfo{IDCard: "440000********1234", PersonName: "测试居民"}, + ManageInfo: contract.ManageInfo{OperateUser: "DOC-1"}, + HealthRecord: contract.HealthRecordCreateInfo{CheckID: "CHECK-STABLE", IDCard: "440000********1234", ManaUnitID: "UNIT-1", ManaDoctorID: "DOC-1"}, + } +} + +func matchingTarget() contract.HealthRecord { + return contract.HealthRecord{IDCard: "440000********1234", PhrID: "PHR-EXISTING", EmpiID: "EMPI-1", ManaUnitID: "UNIT-1", ManaDoctorID: "DOC-1", Status: "1"} +} + +func targetWith(change func(*contract.HealthRecord)) contract.HealthRecord { + r := matchingTarget() + change(&r) + return r +} + +type fakeHealthRecordClient struct { + rows []contract.FindHealthRecord + findResult, saveResult osi.Result + findErr, saveErr error + saved contract.HealthRecordSaveResult + findCalls, createCalls, updateCalls int + updateReq contract.HealthRecordCreate +} + +func (f *fakeHealthRecordClient) FindHealthRecord(context.Context, osi.FindHealthRecordQuery) ([]contract.FindHealthRecord, osi.Result, error) { + f.findCalls++ + return f.rows, f.findResult, f.findErr +} +func (f *fakeHealthRecordClient) CreateHealthRecord(context.Context, contract.HealthRecordCreate) (contract.HealthRecordSaveResult, osi.Result, error) { + f.createCalls++ + return f.saved, f.saveResult, f.saveErr +} +func (f *fakeHealthRecordClient) UpdateHealthRecord(_ context.Context, req contract.HealthRecordCreate) (contract.HealthRecordSaveResult, osi.Result, error) { + f.updateCalls++ + f.updateReq = req + return f.saved, f.saveResult, f.saveErr +} + +type fakeIdempotencyStore struct { + completed, inProgress, marked, released bool + completeErr error +} + +func (f *fakeIdempotencyStore) Acquire(context.Context, string) (IdempotencyLease, error) { + if f.completed { + return IdempotencyLease{State: IdempotencyCompleted}, nil + } + if f.inProgress { + return IdempotencyLease{State: IdempotencyInProgress}, nil + } + return IdempotencyLease{State: IdempotencyAcquired, Token: "lease-1"}, nil +} +func (f *fakeIdempotencyStore) Complete(_ context.Context, _, token string) error { + if token != "lease-1" { + return errors.New("stale lease") + } + f.marked = true + return f.completeErr +} +func (f *fakeIdempotencyStore) Release(_ context.Context, _, token string) error { + if token != "lease-1" { + return errors.New("stale lease") + } + f.released = true + return nil +} + +type fakeReportSink struct{ events []UpsertEvent } + +func (f *fakeReportSink) Publish(_ context.Context, event UpsertEvent) error { + f.events = append(f.events, event) + return nil +} + +type fakeStatusWriter struct { + updates []PHISStatusUpdate + err error +} + +func (f *fakeStatusWriter) WriteStatus(_ context.Context, update PHISStatusUpdate) error { + f.updates = append(f.updates, update) + return f.err +} + +func assertNotifications(t *testing.T, reports *fakeReportSink, statuses *fakeStatusWriter, status Status, action Action) { + t.Helper() + if len(reports.events) != 1 || reports.events[0].Status != status || reports.events[0].Action != action { + t.Fatalf("reports=%#v", reports.events) + } + if len(statuses.updates) != 1 || statuses.updates[0].Status != status { + t.Fatalf("statuses=%#v", statuses.updates) + } +} diff --git a/progress.md b/progress.md index aa54b07..ca5d195 100644 --- a/progress.md +++ b/progress.md @@ -350,3 +350,14 @@ - 安全：真实 PHIS payload 未复制入仓库；fixture 使用占位身份信息和假凭据，生产 DTO 不声明 `sxtAccount/sxtPassword`。 - 验证：`go test ./source -count=1`、`go test ./contract ./mapping -count=1`、`go test ./cache ./osi ./mapping -count=1`、`go test ./... -count=1`、`go vet ./...` 均通过。 - 下一步：T-213 健康档案 query-first upsert 应用编排。 + +## 2026-07-16 T-213 健康档案 query-first upsert 应用编排 + +- 状态：DONE。 +- RED：`go test ./pipeline -count=1` 初次因 `HealthRecordUpsertService`、状态/动作、事件和注入端口均不存在而编译失败。 +- GREEN：新增 `pipeline/health_record_upsert.go`；按身份证查询后仅在明确 0 条时创建，单条记录通过身份证、状态、机构、责任医生和 `phrId` 校验后更新，多条及不安全目标转 `manual_review`；查询/创建/更新失败按 `osi.Result.Retryable` 分类且不互相回退。 +- 幂等：键为 `sha256(checkId + "|" + idCard)`；接口采用带 owner token 的原子 `Acquire/Complete/Release`，并发处理中直接 `retry/skip`，完成/释放校验 token，写入失败释放租约，明确成功后完成租约。 +- 更新语义：查询目标单独建模 `UpdateTarget`，更新请求仅合并平台 `phrId`，源侧稳定 `checkId` 不被查询记录覆盖。 +- 安全与边界：普通报告事件只含哈希 `trace_id`、业务状态、serviceId、响应码及脱敏 `phrId`，原始源标识只交给受控 PHIS 回写端口；`data:null`/缺失不得视为 0 条；幂等完成失败不发送 `done`，副作用失败通过 `NotificationError` 区分租约/报告/PHIS 回写错误并只暴露失败端补偿数据。持久化、重试调度、HTTP 和 PHIS 真实回写不在 T-213 实现。 +- 验证：`go test ./pipeline -count=1`、`go test ./... -count=1`、`go build ./...`、`go vet ./...`、`git diff --check` 均通过。`bash ./init.sh` 在当前 PowerShell 调起的 Bash 环境失败（`go: command not found`），未误报通过；其内部等价 Go 命令已在 PowerShell 单独验证。 +- 下一步：T-204 `POST /api/health-record/upsert` HTTP 适配；T-215 继续等待安全测试档案和写入授权。 diff --git a/tasks.md b/tasks.md index 5f8ccf1..08d348a 100644 --- a/tasks.md +++ b/tasks.md @@ -93,7 +93,7 @@ | ID | 任务 | 依赖 | 验收要点 | 状态 | | --- | --- | --- | --- | --- | | T-212 | PHIS 健康档案真实结构建模与 PHIS→CHIS 转换器 | T-202, T-103 | 建模 PHIS `data.doctor/data.record/archId/businessId/empiId/phrId`；先确认稳定源主键语义并落 ADR；完整转换 JKDA 主体、既往史和生活环境；校验必填/长度/日期/码表；脱敏 fixture 回归；按档案责任医生生成请求操作上下文 | DONE | -| T-213 | 健康档案 upsert 应用编排（外部能力均接口注入） | T-212, T-203, T-206 | 按身份证查询 CHIS：0 条创建、1 条合规档案更新、多条/跨机构/不可更新状态转人工；查询失败不得降级创建；用假 OSI/幂等/report/PHIS 回写实现验收，不在本任务实现持久化 | TODO | +| T-213 | 健康档案 upsert 应用编排（外部能力均接口注入） | T-212, T-203, T-206 | 按身份证查询 CHIS：0 条创建、1 条合规档案更新、多条/跨机构/不可更新状态转人工；查询失败不得降级创建；用假 OSI/幂等/report/PHIS 回写实现验收，不在本任务实现持久化 | DONE | | T-215 | JKDA 健康档案真实 create/update 验收 | T-212, T-213 | 经明确授权的安全测试档案完成 create→query 回查→update→query 回查；确认 `phrId`、checkId、更新目标标识、`addressNumber`、责任医生/机构及重复提交语义，并回填 docs/03 §8、docs/04 §10 | BLOCKED（待写入授权和可写测试档案） | ### T-212 需求与解决方案 @@ -134,6 +134,8 @@ 9. PHIS 真实回写由 T-214 实现；内部状态与 PHIS 实际支持状态分开建模，由适配器完成映射，不能预设 PHIS 原生支持 `manual_review`。 10. T-213 提供可被 HTTP handler 和未来 `deliver-worker` 复用的单一应用服务；T-204 只做 HTTP 适配，不复制 upsert 逻辑。 +**T-213 落地结论（2026-07-16）**：`pipeline.HealthRecordUpsertService` 已实现 query-first 编排；幂等接口采用带 owner token 的原子 `Acquire/Complete/Release`，并发处理中返回 `retry/skip`，避免两个 worker 同时创建和过期 worker 释放新租约。JKDA 查询只有明确返回非 nil 空数组才创建，`data:null`/缺失拒绝写入。单条更新会单独建模 `UpdateTarget`，仅合并查询返回的 `phrId`，保留源侧稳定 `checkId`；状态、机构、责任医生、身份证或 `phrId` 不安全时进入 `manual_review`。持久化、HTTP 入口和 PHIS 真实状态适配仍分别归阶段 3、T-204、T-214。 + ## 里程碑 - M1 = T-005：Go 客户端与平台真实握手成功（签名/信封/判码全对）。