fix(health): 修复审计发现的 10 个 CRITICAL 问题

权限与安全: - 为全部 51 个 handler 端点添加 require_permission 权限检查 - 修复 CAS 预约操作中 doctor_id 为 None 时使用 Uuid::nil() 的问题状态机修复: - 预约初始状态从 "scheduled" 改为 "pending"（匹配设计规格） - 排班状态从 "active" 改为 "enabled" - 咨询会话添加 waiting→active 自动触发（首条消息时） - 新增 create_session 端点和 DTO 数据完整性: - doctor_profile 表添加 name 列（entity + migration + service） - lab_report/health_trend 的 json 列改为 json_binary（支持 GIN 索引） - 添加关键索引：patient.id_number UNIQUE、patient_tag UNIQUE、 doctor_schedule 唯一排班槽位、health_trend、doctor_profile.name - 随访记录完成后自动检查 next_follow_up_date 创建后续任务事件总线: - 实现 10 种核心事件发布（patient/appointment/follow_up/consultation/lab_report） - 实现 workflow.task.completed 和 message.sent 事件订阅框架种子数据: - 实现 seed_tenant_health（8 个默认患者标签） - 实现 soft_delete_tenant_data（16 张表级联软删除）
2026-04-23 23:25:53 +08:00
parent d6678d001e
commit 2e9eb55f2c
18 changed files with 423 additions and 31 deletions
--- a/crates/erp-health/src/service/appointment_service.rs
+++ b/crates/erp-health/src/service/appointment_service.rs
@@ -1,6 +1,7 @@
 //! 预约排班 Service — 预约CRUD、排班管理、日历视图、原子CAS预约

 use chrono::Utc;
+use erp_core::events::DomainEvent;
 use sea_orm::entity::prelude::*;
 use sea_orm::{ActiveValue::Set, Condition, QueryOrder, QuerySelect};
 use uuid::Uuid;
@@ -74,7 +75,7 @@ pub async fn create_appointment(
        )
        .col_expr(doctor_schedule::Column::UpdatedAt, Expr::value(Utc::now()))
        .filter(doctor_schedule::Column::TenantId.eq(tenant_id))
-        .filter(doctor_schedule::Column::DoctorId.eq(req.doctor_id.unwrap_or_default()))
+        .filter(doctor_schedule::Column::DoctorId.eq(req.doctor_id.ok_or(HealthError::Validation("doctor_id is required".to_string()))?))
        .filter(doctor_schedule::Column::ScheduleDate.eq(req.appointment_date))
        .filter(doctor_schedule::Column::StartTime.eq(req.start_time))
        .filter(
@@ -102,7 +103,7 @@ pub async fn create_appointment(
        appointment_date: Set(req.appointment_date),
        start_time: Set(req.start_time),
        end_time: Set(req.end_time),
-        status: Set("scheduled".to_string()),
+        status: Set("pending".to_string()),
        cancel_reason: Set(None),
        notes: Set(req.notes),
        created_at: Set(now),
@@ -113,6 +114,14 @@ pub async fn create_appointment(
        version: Set(1),
    };
    let m = active.insert(&state.db).await?;
+
+    let event = DomainEvent::new(
+        "appointment.created",
+        tenant_id,
+        serde_json::json!({ "appointment_id": m.id, "patient_id": m.patient_id, "status": m.status }),
+    );
+    state.event_bus.publish(event, &state.db).await;
+
    Ok(AppointmentResp {
        id: m.id, patient_id: m.patient_id, doctor_id: m.doctor_id,
        appointment_type: m.appointment_type, appointment_date: m.appointment_date,
@@ -140,7 +149,7 @@ pub async fn update_appointment_status(

    // 状态机校验
    let valid = match (model.status.as_str(), req.status.as_str()) {
-        ("scheduled", "confirmed" | "cancelled") => true,
+        ("pending", "confirmed" | "cancelled") => true,
        ("confirmed", "completed" | "no_show" | "cancelled") => true,
        _ => false,
    };
@@ -179,6 +188,15 @@ pub async fn update_appointment_status(
    active.version = Set(next_ver);

    let m = active.update(&state.db).await?;
+
+    let event_type = format!("appointment.{}", m.status);
+    let event = DomainEvent::new(
+        event_type,
+        tenant_id,
+        serde_json::json!({ "appointment_id": m.id, "patient_id": m.patient_id, "status": m.status }),
+    );
+    state.event_bus.publish(event, &state.db).await;
+
    Ok(AppointmentResp {
        id: m.id, patient_id: m.patient_id, doctor_id: m.doctor_id,
        appointment_type: m.appointment_type, appointment_date: m.appointment_date,
@@ -246,7 +264,7 @@ pub async fn create_schedule(
        end_time: Set(req.end_time),
        max_appointments: Set(req.max_appointments),
        current_appointments: Set(0),
-        status: Set("active".to_string()),
+        status: Set("enabled".to_string()),
        created_at: Set(now),
        updated_at: Set(now),
        created_by: Set(operator_id),
--- a/crates/erp-health/src/service/consultation_service.rs
+++ b/crates/erp-health/src/service/consultation_service.rs
@@ -1,8 +1,9 @@
 //! 咨询管理 Service — 会话管理、消息收发、会话关闭、导出

 use chrono::Utc;
+use erp_core::events::DomainEvent;
 use sea_orm::entity::prelude::*;
-use sea_orm::{ActiveValue::Set, QueryOrder, QuerySelect};
+use sea_orm::{ActiveValue::Set, Condition, QueryOrder, QuerySelect};
 use uuid::Uuid;

 use erp_core::error::check_version;
@@ -17,6 +18,49 @@ use crate::state::HealthState;
 // 咨询会话
 // ---------------------------------------------------------------------------

+pub async fn create_session(
+    state: &HealthState,
+    tenant_id: Uuid,
+    operator_id: Option<Uuid>,
+    req: CreateSessionReq,
+) -> HealthResult<SessionResp> {
+    let now = Utc::now();
+    let active = consultation_session::ActiveModel {
+        id: Set(Uuid::now_v7()),
+        tenant_id: Set(tenant_id),
+        patient_id: Set(req.patient_id),
+        doctor_id: Set(req.doctor_id),
+        consultation_type: Set(req.consultation_type.unwrap_or_else(|| "text".to_string())),
+        status: Set("waiting".to_string()),
+        last_message_at: Set(None),
+        unread_count_patient: Set(0),
+        unread_count_doctor: Set(0),
+        created_at: Set(now),
+        updated_at: Set(now),
+        created_by: Set(operator_id),
+        updated_by: Set(operator_id),
+        deleted_at: Set(None),
+        version: Set(1),
+    };
+    let m = active.insert(&state.db).await?;
+
+    let event = DomainEvent::new(
+        "consultation.opened",
+        tenant_id,
+        serde_json::json!({ "session_id": m.id, "patient_id": m.patient_id }),
+    );
+    state.event_bus.publish(event, &state.db).await;
+
+    Ok(SessionResp {
+        id: m.id, patient_id: m.patient_id, doctor_id: m.doctor_id,
+        consultation_type: m.consultation_type, status: m.status,
+        last_message_at: m.last_message_at,
+        unread_count_patient: m.unread_count_patient,
+        unread_count_doctor: m.unread_count_doctor,
+        created_at: m.created_at,
+    })
+}
+
 pub async fn list_sessions(
    state: &HealthState,
    tenant_id: Uuid,
@@ -84,6 +128,14 @@ pub async fn close_session(
    active.version = Set(next_ver);

    let m = active.update(&state.db).await?;
+
+    let event = DomainEvent::new(
+        "consultation.closed",
+        tenant_id,
+        serde_json::json!({ "session_id": m.id, "patient_id": m.patient_id }),
+    );
+    state.event_bus.publish(event, &state.db).await;
+
    Ok(SessionResp {
        id: m.id, patient_id: m.patient_id, doctor_id: m.doctor_id,
        consultation_type: m.consultation_type, status: m.status,
@@ -138,7 +190,7 @@ pub async fn list_messages(
    let limit = page_size.min(100);
    let offset = page.saturating_sub(1) * limit;

-    let mut query = consultation_message::Entity::find()
+    let query = consultation_message::Entity::find()
        .filter(consultation_message::Column::TenantId.eq(tenant_id))
        .filter(consultation_message::Column::SessionId.eq(session_id))
        .filter(consultation_message::Column::DeletedAt.is_null());
@@ -167,18 +219,23 @@ pub async fn create_message(
    operator_id: Option<Uuid>,
    req: CreateMessageReq,
 ) -> HealthResult<MessageResp> {
-    // 校验会话存在且状态为 active
+    // 校验会话存在且状态为 active 或 waiting
    let session = consultation_session::Entity::find()
        .filter(consultation_session::Column::Id.eq(req.session_id))
        .filter(consultation_session::Column::TenantId.eq(tenant_id))
        .filter(consultation_session::Column::DeletedAt.is_null())
-        .filter(consultation_session::Column::Status.eq("active"))
+        .filter(
+            Condition::any()
+                .add(consultation_session::Column::Status.eq("active"))
+                .add(consultation_session::Column::Status.eq("waiting")),
+        )
        .one(&state.db)
        .await?
        .ok_or(HealthError::ConsultationNotFound)?;

    let now = Utc::now();
    let is_patient = req.sender_role == "patient";
+    let should_activate = session.status == "waiting";

    // 创建消息
    let active = consultation_message::ActiveModel {
@@ -199,9 +256,12 @@ pub async fn create_message(
    };
    let m = active.insert(&state.db).await?;

-    // 更新会话的 last_message_at 和未读计数
+    // 更新会话的 last_message_at 和未读计数，waiting→active 自动触发
    let mut session_active: consultation_session::ActiveModel = session.into();
    session_active.last_message_at = Set(Some(now));
+    if should_activate {
+        session_active.status = Set("active".to_string());
+    }
    // 根据发送者角色更新对方的 unread_count
    if is_patient {
        session_active.unread_count_doctor = Set(session_active.unread_count_doctor.unwrap() + 1);
--- a/crates/erp-health/src/service/doctor_service.rs
+++ b/crates/erp-health/src/service/doctor_service.rs
@@ -30,10 +30,10 @@ pub async fn list_doctors(
        .filter(doctor_profile::Column::DeletedAt.is_null());

    if let Some(ref s) = search {
-        // doctor_profile 没有 name 字段，按 license_number/department 模糊搜索
        let pattern = format!("%{}%", s);
        query = query.filter(
            Condition::any()
+                .add(doctor_profile::Column::Name.contains(&pattern))
                .add(doctor_profile::Column::LicenseNumber.contains(&pattern))
                .add(doctor_profile::Column::Department.contains(&pattern))
                .add(doctor_profile::Column::Specialty.contains(&pattern)),
@@ -79,6 +79,7 @@ pub async fn create_doctor(
        id: Set(id),
        tenant_id: Set(tenant_id),
        user_id: Set(req.user_id),
+        name: Set(req.name),
        department: Set(req.department),
        title: Set(req.title),
        specialty: Set(req.specialty),
@@ -119,8 +120,7 @@ pub async fn update_doctor(
        .map_err(|_| HealthError::VersionMismatch)?;

    let mut active: doctor_profile::ActiveModel = model.into();
-    // doctor_profile 没有 name 字段，但 handler DTO 有 name
-    // name 需要通过 user_id 关联到 erp-auth 的 users 表，此处暂不处理
+    if let Some(v) = req.name { active.name = Set(v); }
    if let Some(v) = req.department { active.department = Set(Some(v)); }
    if let Some(v) = req.title { active.title = Set(Some(v)); }
    if let Some(v) = req.specialty { active.specialty = Set(Some(v)); }
@@ -170,7 +170,7 @@ fn model_to_resp(m: doctor_profile::Model) -> DoctorResp {
    DoctorResp {
        id: m.id,
        user_id: m.user_id,
-        name: m.user_id.map(|_| "".to_string()).unwrap_or_default(),
+        name: m.name,
        department: m.department,
        title: m.title,
        specialty: m.specialty,
--- a/crates/erp-health/src/service/follow_up_service.rs
+++ b/crates/erp-health/src/service/follow_up_service.rs
@@ -1,6 +1,7 @@
 //! 随访管理 Service — 随访任务CRUD、随访记录、状态流转

 use chrono::Utc;
+use erp_core::events::DomainEvent;
 use sea_orm::entity::prelude::*;
 use sea_orm::{ActiveValue::Set, QueryOrder, QuerySelect};
 use uuid::Uuid;
@@ -82,6 +83,14 @@ pub async fn create_task(
        version: Set(1),
    };
    let m = active.insert(&state.db).await?;
+
+    let event = DomainEvent::new(
+        "follow_up.created",
+        tenant_id,
+        serde_json::json!({ "task_id": m.id, "patient_id": m.patient_id }),
+    );
+    state.event_bus.publish(event, &state.db).await;
+
    Ok(FollowUpTaskResp {
        id: m.id, patient_id: m.patient_id, assigned_to: m.assigned_to,
        follow_up_type: m.follow_up_type, planned_date: m.planned_date,
@@ -197,12 +206,44 @@ pub async fn create_record(
    let record = record_active.insert(&state.db).await?;

    let mut task_active: follow_up_task::ActiveModel = task.into();
+    let task_patient_id = task_active.patient_id.clone().unwrap();
+    let task_assigned_to = task_active.assigned_to.clone().unwrap();
+    let task_follow_up_type = task_active.follow_up_type.clone().unwrap();
    task_active.status = Set("completed".to_string());
    task_active.updated_at = Set(now);
    task_active.updated_by = Set(operator_id);
    task_active.version = Set(task_active.version.unwrap() + 1);
    task_active.update(&state.db).await?;

+    // 当 next_follow_up_date 不为空时，自动创建后续随访任务
+    if let Some(next_date) = req.next_follow_up_date {
+        let new_task = follow_up_task::ActiveModel {
+            id: Set(Uuid::now_v7()),
+            tenant_id: Set(tenant_id),
+            patient_id: Set(task_patient_id),
+            assigned_to: Set(task_assigned_to),
+            follow_up_type: Set(task_follow_up_type),
+            planned_date: Set(next_date),
+            status: Set("pending".to_string()),
+            content_template: Set(None),
+            related_appointment_id: Set(None),
+            created_at: Set(now),
+            updated_at: Set(now),
+            created_by: Set(operator_id),
+            updated_by: Set(operator_id),
+            deleted_at: Set(None),
+            version: Set(1),
+        };
+        new_task.insert(&state.db).await?;
+    }
+
+    let event = DomainEvent::new(
+        "follow_up.completed",
+        tenant_id,
+        serde_json::json!({ "task_id": record.task_id, "patient_id": task_patient_id }),
+    );
+    state.event_bus.publish(event, &state.db).await;
+
    Ok(FollowUpRecordResp {
        id: record.id, task_id: record.task_id, executed_by: record.executed_by,
        executed_date: record.executed_date, result: record.result,
--- a/crates/erp-health/src/service/health_data_service.rs
+++ b/crates/erp-health/src/service/health_data_service.rs
@@ -1,6 +1,7 @@
 //! 健康数据 Service — 体征记录、化验报告、体检记录、趋势分析

 use chrono::Utc;
+use erp_core::events::DomainEvent;
 use sea_orm::entity::prelude::*;
 use sea_orm::{ActiveValue::Set, QueryOrder, QuerySelect};
 use uuid::Uuid;
@@ -27,7 +28,7 @@ pub async fn list_vital_signs(
    let limit = page_size.min(100);
    let offset = page.saturating_sub(1) * limit;

-    let mut query = vital_signs::Entity::find()
+    let query = vital_signs::Entity::find()
        .filter(vital_signs::Column::TenantId.eq(tenant_id))
        .filter(vital_signs::Column::PatientId.eq(patient_id))
        .filter(vital_signs::Column::DeletedAt.is_null());
@@ -191,7 +192,7 @@ pub async fn list_lab_reports(
    let limit = page_size.min(100);
    let offset = page.saturating_sub(1) * limit;

-    let mut query = lab_report::Entity::find()
+    let query = lab_report::Entity::find()
        .filter(lab_report::Column::TenantId.eq(tenant_id))
        .filter(lab_report::Column::PatientId.eq(patient_id))
        .filter(lab_report::Column::DeletedAt.is_null());
@@ -240,6 +241,14 @@ pub async fn create_lab_report(
        version: Set(1),
    };
    let m = active.insert(&state.db).await?;
+
+    let event = DomainEvent::new(
+        "lab_report.uploaded",
+        tenant_id,
+        serde_json::json!({ "report_id": m.id, "patient_id": m.patient_id, "report_type": m.report_type }),
+    );
+    state.event_bus.publish(event, &state.db).await;
+
    Ok(LabReportResp {
        id: m.id, patient_id: m.patient_id, report_date: m.report_date,
        report_type: m.report_type, indicators: m.indicators,
@@ -322,7 +331,7 @@ pub async fn list_health_records(
    let limit = page_size.min(100);
    let offset = page.saturating_sub(1) * limit;

-    let mut query = health_record::Entity::find()
+    let query = health_record::Entity::find()
        .filter(health_record::Column::TenantId.eq(tenant_id))
        .filter(health_record::Column::PatientId.eq(patient_id))
        .filter(health_record::Column::DeletedAt.is_null());
@@ -455,7 +464,7 @@ pub async fn list_trends(
    let limit = page_size.min(100);
    let offset = page.saturating_sub(1) * limit;

-    let mut query = health_trend::Entity::find()
+    let query = health_trend::Entity::find()
        .filter(health_trend::Column::TenantId.eq(tenant_id))
        .filter(health_trend::Column::PatientId.eq(patient_id))
        .filter(health_trend::Column::DeletedAt.is_null());
--- a/crates/erp-health/src/service/patient_service.rs
+++ b/crates/erp-health/src/service/patient_service.rs
@@ -1,6 +1,7 @@
 //! 患者管理 Service — CRUD、家庭成员、标签、医生关联、健康摘要

 use chrono::Utc;
+use erp_core::events::DomainEvent;
 use sea_orm::entity::prelude::*;
 use sea_orm::{ActiveValue::Set, Condition, QueryOrder, QuerySelect};
 use uuid::Uuid;
@@ -122,6 +123,14 @@ pub async fn create_patient(
    };

    let model = active.insert(&state.db).await?;
+
+    let event = DomainEvent::new(
+        "patient.created",
+        tenant_id,
+        serde_json::json!({ "patient_id": model.id, "name": model.name }),
+    );
+    state.event_bus.publish(event, &state.db).await;
+
    Ok(model_to_resp(model))
 }

@@ -167,6 +176,14 @@ pub async fn update_patient(
    active.version = Set(next_ver);

    let updated = active.update(&state.db).await?;
+
+    let event = DomainEvent::new(
+        "patient.updated",
+        tenant_id,
+        serde_json::json!({ "patient_id": updated.id }),
+    );
+    state.event_bus.publish(event, &state.db).await;
+
    Ok(model_to_resp(updated))
 }

@@ -277,7 +294,7 @@ pub async fn get_health_summary(
    let upcoming = appointment::Entity::find()
        .filter(appointment::Column::TenantId.eq(tenant_id))
        .filter(appointment::Column::PatientId.eq(patient_id))
-        .filter(appointment::Column::Status.eq("scheduled"))
+        .filter(appointment::Column::Status.eq("pending"))
        .filter(appointment::Column::DeletedAt.is_null())
        .count(&state.db)
        .await?;
--- a/crates/erp-health/src/service/seed.rs
+++ b/crates/erp-health/src/service/seed.rs
@@ -1,22 +1,95 @@
-//! 租户初始化种子数据 — 创建默认标签、默认排班模板等
+//! 租户初始化种子数据 — 创建默认标签

-use sea_orm::DatabaseConnection;
+use chrono::Utc;
+use sea_orm::{ActiveModelTrait, ActiveValue::Set, ConnectionTrait, DatabaseConnection};
 use uuid::Uuid;

+use crate::entity::patient_tag;
+
+const DEFAULT_TAGS: &[(&str, &str, &str)] = &[
+    ("高血压", "#E74C3C", "高血压患者标签"),
+    ("糖尿病", "#3498DB", "糖尿病患者标签"),
+    ("心脏病", "#9B59B6", "心脏病患者标签"),
+    ("过敏体质", "#F39C12", "过敏体质患者标签"),
+    ("老年患者", "#1ABC9C", "65岁以上老年患者"),
+    ("孕产妇", "#E91E63", "孕产期健康管理"),
+    ("慢性病", "#607D8B", "慢性病长期随访"),
+    ("术后随访", "#795548", "手术后随访管理"),
+];
+
 /// 初始化租户健康模块默认数据
 pub async fn seed_tenant_health(
-    _db: &DatabaseConnection,
+    db: &DatabaseConnection,
    tenant_id: Uuid,
 ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    tracing::info!(tenant_id = %tenant_id, "Seeding health module default data");
+
+    let now = Utc::now();
+    for (name, color, description) in DEFAULT_TAGS {
+        let active = patient_tag::ActiveModel {
+            id: Set(Uuid::now_v7()),
+            tenant_id: Set(tenant_id),
+            name: Set(name.to_string()),
+            color: Set(Some(color.to_string())),
+            description: Set(Some(description.to_string())),
+            is_system: Set(true),
+            created_at: Set(now),
+            updated_at: Set(now),
+            created_by: Set(None),
+            updated_by: Set(None),
+            deleted_at: Set(None),
+            version: Set(1),
+        };
+        active.insert(db).await?;
+    }
+
+    tracing::info!(tenant_id = %tenant_id, "Health module default data seeded successfully");
    Ok(())
 }

 /// 软删除该租户下所有健康模块数据
 pub async fn soft_delete_tenant_data(
-    _db: &DatabaseConnection,
+    db: &DatabaseConnection,
    tenant_id: Uuid,
 ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    tracing::info!(tenant_id = %tenant_id, "Soft-deleting health module data for tenant");
+
+    let now = Utc::now();
+
+    let tables_to_soft_delete: Vec<&str> = vec![
+        "consultation_message",
+        "consultation_session",
+        "follow_up_record",
+        "follow_up_task",
+        "appointment",
+        "doctor_schedule",
+        "health_trend",
+        "lab_report",
+        "health_record",
+        "vital_signs",
+        "patient_doctor_relation",
+        "patient_tag_relation",
+        "patient_family_member",
+        "patient",
+        "patient_tag",
+        "doctor_profile",
+    ];
+
+    for table in tables_to_soft_delete {
+        let sql = format!(
+            "UPDATE {} SET deleted_at = $1, updated_at = $1 WHERE tenant_id = $2 AND deleted_at IS NULL",
+            table
+        );
+        let stmt = sea_orm::Statement::from_sql_and_values(
+            sea_orm::DatabaseBackend::Postgres,
+            &sql,
+            [now.into(), tenant_id.into()],
+        );
+        if let Err(e) = db.execute(stmt).await {
+            tracing::warn!(table = %table, error = %e, "Failed to soft-delete table");
+        }
+    }
+
+    tracing::info!(tenant_id = %tenant_id, "Health module data soft-deleted");
    Ok(())
 }