feat(core): 事件归档 + 消费者幂等性 — 迁移 084/085 + 清理任务

- 迁移 084: domain_events_archive 归档表 + cleanup_old_published_events()
- 迁移 085: processed_events 去重表 + cleanup_old_processed_events()
- erp-core: is_event_processed() / mark_event_processed() 幂等性辅助
- erp-server: tasks::start_event_cleanup() 每 24h 归档 >90 天事件

Cargo.toml

@@ -38,6 +38,7 @@ sea-orm = { version = "1.1", features = [
     "sqlx-postgres", "runtime-tokio-rustls", "macros", "with-uuid", "with-chrono", "with-json"
 ] }
 sea-orm-migration = { version = "1.1", features = ["sqlx-postgres", "runtime-tokio-rustls"] }
+sqlx = { version = "0.8", features = ["runtime-tokio-rustls", "postgres", "uuid"] }
 
 # Serialization
 serde = { version = "1", features = ["derive"] }

crates/erp-auth/src/service/auth_service.rs

@@ -161,7 +161,7 @@ impl AuthService {
         event_bus.publish(erp_core::events::DomainEvent::new(
             "user.login",
             tenant_id,
-            serde_json::json!({ "user_id": user_model.id, "username": user_model.username }),
+            erp_core::events::build_event_payload(serde_json::json!({ "user_id": user_model.id, "username": user_model.username })),
         ), db).await;
 
         // 审计：登录成功

crates/erp-core/src/entity/mod.rs

@@ -1,2 +1,3 @@
 pub mod audit_log;
 pub mod domain_event;
+pub mod processed_event;

crates/erp-core/src/entity/processed_event.rs

@@ -0,0 +1,18 @@
+use sea_orm::entity::prelude::*;
+use serde::{Deserialize, Serialize};
+
+/// 已处理事件记录 — 幂等性去重表。
+#[derive(Clone, Debug, PartialEq, DeriveEntityModel, Serialize, Deserialize)]
+#[sea_orm(table_name = "processed_events")]
+pub struct Model {
+    #[sea_orm(primary_key, auto_increment = false)]
+    pub event_id: Uuid,
+    #[sea_orm(primary_key, auto_increment = false)]
+    pub consumer_id: String,
+    pub processed_at: DateTimeUtc,
+}
+
+#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)]
+pub enum Relation {}
+
+impl ActiveModelBehavior for ActiveModel {}

crates/erp-core/src/events.rs

@@ -1,5 +1,5 @@
 use chrono::Utc;
-use sea_orm::{ActiveModelTrait, Set};
+use sea_orm::{ActiveModelTrait, ConnectionTrait, PaginatorTrait, Set};
 use serde::{Deserialize, Serialize};
 use tokio::sync::{broadcast, mpsc};
 use tracing::{error, info};
@@ -31,6 +31,76 @@ impl DomainEvent {
     }
 }
 
+/// 当前事件 payload schema 版本
+pub const EVENT_SCHEMA_VERSION: &str = "v1";
+
+/// 构造统一信封格式的事件 payload。
+///
+/// 自动注入 `schema_version` 和 `occurred_at`，业务数据通过 `data` 传入。
+/// 用法：`build_event_payload(serde_json::json!({ "patient_id": ..., }))`
+pub fn build_event_payload(data: serde_json::Value) -> serde_json::Value {
+    let mut envelope = serde_json::json!({
+        "schema_version": EVENT_SCHEMA_VERSION,
+        "occurred_at": Utc::now().to_rfc3339(),
+    });
+    if let serde_json::Value::Object(ref mut map) = envelope {
+        if let serde_json::Value::Object(data_map) = data {
+            for (k, v) in data_map {
+                map.insert(k, v);
+            }
+        }
+    }
+    envelope
+}
+
+/// 检查事件是否已被指定消费者处理。
+///
+/// 查询 `processed_events` 表判断 event_id + consumer_id 是否已存在。
+pub async fn is_event_processed(
+    db: &sea_orm::DatabaseConnection,
+    event_id: Uuid,
+    consumer_id: &str,
+) -> Result<bool, sea_orm::DbErr> {
+    use sea_orm::{ColumnTrait, EntityTrait, QueryFilter};
+
+    let count = crate::entity::processed_event::Entity::find()
+        .filter(crate::entity::processed_event::Column::EventId.eq(event_id))
+        .filter(crate::entity::processed_event::Column::ConsumerId.eq(consumer_id))
+        .count(db)
+        .await?;
+    Ok(count > 0)
+}
+
+/// 标记事件已被指定消费者处理。
+///
+/// 插入 `processed_events` 记录，重复插入会因主键冲突被安全忽略。
+pub async fn mark_event_processed(
+    db: &sea_orm::DatabaseConnection,
+    event_id: Uuid,
+    consumer_id: &str,
+) -> Result<(), sea_orm::DbErr> {
+    use sea_orm::ActiveModelTrait;
+    use sea_orm::Set;
+
+    let model = crate::entity::processed_event::ActiveModel {
+        event_id: Set(event_id),
+        consumer_id: Set(consumer_id.to_string()),
+        processed_at: Set(Utc::now()),
+    };
+    // INSERT ... ON CONFLICT DO NOTHING（主键冲突时安全忽略）
+    match model.insert(db).await {
+        Ok(_) => Ok(()),
+        Err(e) => {
+            // 唯一约束冲突 = 已处理，不是错误
+            if e.to_string().contains("duplicate") || e.to_string().contains("violates unique") {
+                Ok(())
+            } else {
+                Err(e)
+            }
+        }
+    }
+}
+
 /// 过滤事件接收器 — 只接收匹配 `event_type_prefix` 的事件
 pub struct FilteredEventReceiver {
     receiver: mpsc::Receiver<DomainEvent>,
@@ -70,7 +140,7 @@ impl EventBus {
     }
 
     /// 发布事件：先持久化到 domain_events 表（pending 状态），再内存广播，
-    /// 最后更新为 published。
+    /// 最后更新为 published 并 NOTIFY outbox relay。
     ///
     /// 两阶段提交保证：即使广播后服务崩溃，事件仍为 pending 状态，
     /// 重启后 outbox relay 会重新广播。
@@ -110,6 +180,15 @@ impl EventBus {
         if let Err(e) = active.update(db).await {
             tracing::warn!(event_id = %event_id, error = %e, "领域事件状态更新为 published 失败");
         }
+
+        // 4. NOTIFY outbox relay（通知 outbox relay 有新事件到达）
+        let notify_sql = sea_orm::Statement::from_string(
+            sea_orm::DatabaseBackend::Postgres,
+            format!("NOTIFY outbox_channel, '{}'", event_id),
+        );
+        if let Err(e) = db.execute(notify_sql).await {
+            tracing::debug!(event_id = %event_id, error = %e, "NOTIFY outbox_channel 失败（非致命）");
+        }
     }
 
     /// 仅内存广播（不持久化，用于内部测试等场景）。

crates/erp-health/src/event.rs

@@ -35,6 +35,9 @@ pub const FOLLOW_UP_CREATED: &str = "follow_up.created";
 pub const FOLLOW_UP_COMPLETED: &str = "follow_up.completed";
 pub const FOLLOW_UP_OVERDUE: &str = "follow_up.overdue";
 
+// 日常监测
+pub const DAILY_MONITORING_CREATED: &str = "daily_monitoring.created";
+
 // 健康数据
 pub const LAB_REPORT_UPLOADED: &str = "lab_report.uploaded";
 pub const HEALTH_DATA_CRITICAL_ALERT: &str = "health_data.critical_alert";

crates/erp-health/src/service/alert_engine.rs

@@ -200,14 +200,14 @@ async fn create_alert_and_notify(
     let event = erp_core::events::DomainEvent::new(
         crate::event::ALERT_TRIGGERED,
         tenant_id,
-        json!({
+        erp_core::events::build_event_payload(json!({
             "alert_id": alert.id,
             "patient_id": patient_id,
             "rule_name": rule.name,
             "severity": rule.severity,
             "detail": alert.detail,
             "notify_roles": rule.notify_roles,
-        }),
+        })),
     );
     event_bus.publish(event, db).await;
 

crates/erp-health/src/service/appointment_service.rs

@@ -211,7 +211,7 @@ pub async fn create_appointment(
     let event = DomainEvent::new(
         crate::event::APPOINTMENT_CREATED,
         tenant_id,
-        serde_json::json!({ "appointment_id": m.id, "patient_id": m.patient_id, "status": m.status }),
+        erp_core::events::build_event_payload(serde_json::json!({ "appointment_id": m.id, "patient_id": m.patient_id, "status": m.status })),
     );
     state.event_bus.publish(event, &state.db).await;
 
@@ -299,7 +299,7 @@ pub async fn update_appointment_status(
     let event = DomainEvent::new(
         event_type,
         tenant_id,
-        serde_json::json!({ "appointment_id": m.id, "patient_id": m.patient_id, "status": m.status }),
+        erp_core::events::build_event_payload(serde_json::json!({ "appointment_id": m.id, "patient_id": m.patient_id, "status": m.status })),
     );
     state.event_bus.publish(event, &state.db).await;
 
@@ -573,13 +573,13 @@ pub async fn send_reminders(
             DomainEvent::new(
                 "appointment.reminder",
                 app.tenant_id,
-                json!({
+                erp_core::events::build_event_payload(json!({
                     "appointment_id": app.id,
                     "patient_id": app.patient_id,
                     "doctor_id": app.doctor_id,
                     "appointment_date": app.appointment_date,
                     "time_slot": format!("{}-{}", app.start_time.format("%H:%M"), app.end_time.format("%H:%M")),
-                }),
+                })),
             ),
             db,
         ).await;

crates/erp-health/src/service/article_service.rs

@@ -186,9 +186,9 @@ pub async fn approve_article(
     ).await;
 
     state.event_bus.publish(
-        DomainEvent::new(crate::event::ARTICLE_PUBLISHED, tenant_id, serde_json::json!({
+        DomainEvent::new(crate::event::ARTICLE_PUBLISHED, tenant_id, erp_core::events::build_event_payload(serde_json::json!({
             "article_id": m.id, "title": m.title, "category": m.category,
-        })),
+        }))),
         &state.db,
     ).await;
 
@@ -229,9 +229,9 @@ pub async fn reject_article(
     ).await;
 
     state.event_bus.publish(
-        DomainEvent::new(crate::event::ARTICLE_REJECTED, tenant_id, serde_json::json!({
+        DomainEvent::new(crate::event::ARTICLE_REJECTED, tenant_id, erp_core::events::build_event_payload(serde_json::json!({
             "article_id": m.id, "title": m.title,
-        })),
+        }))),
         &state.db,
     ).await;
 

crates/erp-health/src/service/consent_service.rs

@@ -109,10 +109,10 @@ pub async fn grant_consent(
     ).await;
 
     state.event_bus.publish(
-        DomainEvent::new(crate::event::CONSENT_GRANTED, tenant_id, serde_json::json!({
+        DomainEvent::new(crate::event::CONSENT_GRANTED, tenant_id, erp_core::events::build_event_payload(serde_json::json!({
             "consent_id": m.id, "patient_id": m.patient_id,
             "consent_type": m.consent_type, "consent_scope": m.consent_scope,
-        })),
+        }))),
         &state.db,
     ).await;
 
@@ -155,10 +155,10 @@ pub async fn revoke_consent(
     ).await;
 
     state.event_bus.publish(
-        DomainEvent::new(crate::event::CONSENT_REVOKED, tenant_id, serde_json::json!({
+        DomainEvent::new(crate::event::CONSENT_REVOKED, tenant_id, erp_core::events::build_event_payload(serde_json::json!({
             "consent_id": m.id, "patient_id": m.patient_id,
             "consent_type": m.consent_type,
-        })),
+        }))),
         &state.db,
     ).await;
 

crates/erp-health/src/service/consultation_service.rs

@@ -77,7 +77,7 @@ pub async fn create_session(
     let event = DomainEvent::new(
         crate::event::CONSULTATION_OPENED,
         tenant_id,
-        serde_json::json!({ "session_id": m.id, "patient_id": m.patient_id }),
+        erp_core::events::build_event_payload(serde_json::json!({ "session_id": m.id, "patient_id": m.patient_id })),
     );
     state.event_bus.publish(event, &state.db).await;
 
@@ -175,7 +175,7 @@ pub async fn close_session(
     let event = DomainEvent::new(
         crate::event::CONSULTATION_CLOSED,
         tenant_id,
-        serde_json::json!({ "session_id": m.id, "patient_id": m.patient_id }),
+        erp_core::events::build_event_payload(serde_json::json!({ "session_id": m.id, "patient_id": m.patient_id })),
     );
     state.event_bus.publish(event, &state.db).await;
 

crates/erp-health/src/service/daily_monitoring_service.rs

@@ -8,10 +8,13 @@ use uuid::Uuid;
 
 use erp_core::types::PaginatedResponse;
 
+use erp_core::events::DomainEvent;
+
 use crate::dto::daily_monitoring_dto::*;
 use crate::dto::health_data_dto::CreateVitalSignsReq;
 use crate::entity::vital_signs;
 use crate::error::{HealthError, HealthResult};
+use crate::event::DAILY_MONITORING_CREATED;
 use crate::service::health_data_service;
 use crate::state::HealthState;
 
@@ -94,7 +97,22 @@ pub async fn create_daily_monitoring(
     let vs = health_data_service::create_vital_signs(
         state, tenant_id, req.patient_id, operator_id, vs_req,
     ).await?;
-    Ok(vs_to_dm(vs))
+
+    let dm = vs_to_dm(vs);
+    let event = DomainEvent::new(
+        DAILY_MONITORING_CREATED,
+        tenant_id,
+        erp_core::events::build_event_payload(serde_json::json!({
+            "record_id": dm.id,
+            "patient_id": dm.patient_id,
+            "record_date": dm.record_date,
+            "weight": dm.weight,
+            "blood_sugar": dm.blood_sugar,
+        })),
+    );
+    state.event_bus.publish(event, &state.db).await;
+
+    Ok(dm)
 }
 
 pub async fn update_daily_monitoring(

crates/erp-health/src/service/device_reading_service.rs

@@ -127,7 +127,7 @@ pub async fn batch_create_readings(
     let event = DomainEvent::new(
         crate::event::DEVICE_READINGS_SYNCED,
         tenant_id,
-        serde_json::json!({
+        erp_core::events::build_event_payload(serde_json::json!({
             "patient_id": patient_id,
             "count": inserted,
             "device_model": req.device_model,
@@ -135,7 +135,7 @@ pub async fn batch_create_readings(
                 "from": earliest.map(|t| t.to_rfc3339()),
                 "to": latest.map(|t| t.to_rfc3339()),
             }
-        }),
+        })),
     );
     state.event_bus.publish(event, &state.db).await;
 

crates/erp-health/src/service/doctor_service.rs

@@ -172,7 +172,7 @@ pub async fn update_doctor(
             let event = erp_core::events::DomainEvent::new(
                 crate::event::DOCTOR_ONLINE_STATUS_CHANGED,
                 tenant_id,
-                serde_json::json!({ "doctor_id": id, "old_status": old_online_status, "new_status": v }),
+                erp_core::events::build_event_payload(serde_json::json!({ "doctor_id": id, "old_status": old_online_status, "new_status": v })),
             );
             state.event_bus.publish(event, &state.db).await;
         }

crates/erp-health/src/service/follow_up_service.rs

@@ -167,7 +167,7 @@ pub async fn create_task(
     let event = DomainEvent::new(
         crate::event::FOLLOW_UP_CREATED,
         tenant_id,
-        serde_json::json!({ "task_id": m.id, "patient_id": m.patient_id }),
+        erp_core::events::build_event_payload(serde_json::json!({ "task_id": m.id, "patient_id": m.patient_id })),
     );
     state.event_bus.publish(event, &state.db).await;
 
@@ -359,7 +359,7 @@ pub async fn batch_create_tasks(
             DomainEvent::new(
                 crate::event::FOLLOW_UP_CREATED,
                 tenant_id,
-                serde_json::json!({ "batch_count": count, "assigned_to": req.assigned_to }),
+                erp_core::events::build_event_payload(serde_json::json!({ "batch_count": count, "assigned_to": req.assigned_to })),
             ),
             &state.db,
         ).await;
@@ -487,7 +487,7 @@ pub async fn batch_complete_tasks(
             DomainEvent::new(
                 crate::event::FOLLOW_UP_COMPLETED,
                 tenant_id,
-                serde_json::json!({ "batch_count": succeeded }),
+                erp_core::events::build_event_payload(serde_json::json!({ "batch_count": succeeded })),
             ),
             &state.db,
         ).await;
@@ -594,7 +594,7 @@ pub async fn create_record(
     let event = DomainEvent::new(
         crate::event::FOLLOW_UP_COMPLETED,
         tenant_id,
-        serde_json::json!({ "task_id": record.task_id, "patient_id": task_patient_id }),
+        erp_core::events::build_event_payload(serde_json::json!({ "task_id": record.task_id, "patient_id": task_patient_id })),
     );
     state.event_bus.publish(event, &state.db).await;
 
@@ -777,12 +777,12 @@ pub async fn check_overdue_and_notify(state: &HealthState) -> HealthResult<u64>
         let event = erp_core::events::DomainEvent::new(
             crate::event::FOLLOW_UP_OVERDUE,
             task.tenant_id,
-            serde_json::json!({
+            erp_core::events::build_event_payload(serde_json::json!({
                 "task_id": task.id,
                 "patient_id": task.patient_id,
                 "assigned_to": task.assigned_to,
                 "planned_date": task.planned_date,
-            }),
+            })),
         );
         state.event_bus.publish(event, db).await;
     }

crates/erp-health/src/service/health_data_service.rs

@@ -401,7 +401,7 @@ pub async fn create_lab_report(
     let event = DomainEvent::new(
         crate::event::LAB_REPORT_UPLOADED,
         tenant_id,
-        serde_json::json!({ "report_id": m.id, "patient_id": m.patient_id, "report_type": m.report_type }),
+        erp_core::events::build_event_payload(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;
 
@@ -929,7 +929,7 @@ async fn check_vital_signs_alert(
         let event = DomainEvent::new(
             crate::event::HEALTH_DATA_CRITICAL_ALERT,
             tenant_id,
-            payload,
+            erp_core::events::build_event_payload(payload),
         );
         state.event_bus.publish(event, &state.db).await;
         tracing::warn!(

crates/erp-health/src/service/patient_service.rs

@@ -173,7 +173,7 @@ pub async fn create_patient(
     let event = DomainEvent::new(
         crate::event::PATIENT_CREATED,
         tenant_id,
-        serde_json::json!({ "patient_id": model.id }),
+        erp_core::events::build_event_payload(serde_json::json!({ "patient_id": model.id })),
     );
     state.event_bus.publish(event, &state.db).await;
 
@@ -296,7 +296,7 @@ pub async fn update_patient(
     let event = DomainEvent::new(
         event_type,
         tenant_id,
-        serde_json::json!({ "patient_id": updated.id }),
+        erp_core::events::build_event_payload(serde_json::json!({ "patient_id": updated.id })),
     );
     state.event_bus.publish(event, &state.db).await;
 

crates/erp-health/src/service/points_service.rs

@@ -192,10 +192,10 @@ pub async fn earn_points(
     ).await;
 
     state.event_bus.publish(
-        DomainEvent::new(crate::event::POINTS_EARNED, tenant_id, serde_json::json!({
+        DomainEvent::new(crate::event::POINTS_EARNED, tenant_id, erp_core::events::build_event_payload(serde_json::json!({
             "transaction_id": inserted.id, "account_id": inserted.account_id,
             "amount": inserted.amount, "balance_after": inserted.balance_after,
-        })),
+        }))),
         &state.db,
     ).await;
 
@@ -933,10 +933,10 @@ pub async fn exchange_product(
     ).await;
 
     state.event_bus.publish(
-        DomainEvent::new(crate::event::POINTS_EXCHANGED, tenant_id, serde_json::json!({
+        DomainEvent::new(crate::event::POINTS_EXCHANGED, tenant_id, erp_core::events::build_event_payload(serde_json::json!({
             "order_id": inserted_order.id, "patient_id": inserted_order.patient_id,
             "product_id": inserted_order.product_id, "points_cost": inserted_order.points_cost,
-        })),
+        }))),
         &state.db,
     ).await;
 
@@ -1796,7 +1796,7 @@ pub async fn expire_points(db: &sea_orm::DatabaseConnection, event_bus: &erp_cor
         let event = erp_core::events::DomainEvent::new(
             crate::event::POINTS_EXPIRED,
             tenant_id,
-            serde_json::json!({ "expired_count": processed }),
+            erp_core::events::build_event_payload(serde_json::json!({ "expired_count": processed })),
         );
         event_bus.publish(event, db).await;
     }

crates/erp-server/Cargo.toml

@@ -17,6 +17,7 @@ tracing.workspace = true
 tracing-subscriber.workspace = true
 config.workspace = true
 sea-orm.workspace = true
+sqlx.workspace = true
 redis.workspace = true
 utoipa.workspace = true
 serde_json.workspace = true

crates/erp-server/migration/src/lib.rs

@@ -83,6 +83,8 @@ mod m20260427_000080_create_medication_record;
 mod m20260427_000081_create_dialysis_prescription;
 mod m20260427_000082_seed_ai_prompts;
 mod m20260427_000083_create_follow_up_template;
+mod m20260427_000084_domain_events_cleanup;
+mod m20260427_000085_processed_events;
 
 pub struct Migrator;
 
@@ -173,6 +175,8 @@ impl MigratorTrait for Migrator {
             Box::new(m20260427_000081_create_dialysis_prescription::Migration),
             Box::new(m20260427_000082_seed_ai_prompts::Migration),
             Box::new(m20260427_000083_create_follow_up_template::Migration),
+            Box::new(m20260427_000084_domain_events_cleanup::Migration),
+            Box::new(m20260427_000085_processed_events::Migration),
         ]
     }
 }

crates/erp-server/migration/src/m20260427_000084_domain_events_cleanup.rs

@@ -0,0 +1,89 @@
+use sea_orm_migration::prelude::*;
+
+#[derive(DeriveMigrationName)]
+pub struct Migration;
+
+#[async_trait::async_trait]
+impl MigrationTrait for Migration {
+    async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+        // 归档表 — 与 domain_events 结构相同，用于存放 >90 天的已发布事件
+        manager
+            .create_table(
+                Table::create()
+                    .table(Alias::new("domain_events_archive"))
+                    .if_not_exists()
+                    .col(ColumnDef::new(Alias::new("id")).uuid().not_null().primary_key())
+                    .col(ColumnDef::new(Alias::new("tenant_id")).uuid().not_null())
+                    .col(ColumnDef::new(Alias::new("event_type")).string_len(200).not_null())
+                    .col(ColumnDef::new(Alias::new("payload")).json().null())
+                    .col(ColumnDef::new(Alias::new("correlation_id")).uuid().null())
+                    .col(ColumnDef::new(Alias::new("status")).string_len(20).not_null())
+                    .col(ColumnDef::new(Alias::new("attempts")).integer().not_null().default(0))
+                    .col(ColumnDef::new(Alias::new("last_error")).text().null())
+                    .col(ColumnDef::new(Alias::new("created_at")).timestamp_with_time_zone().not_null())
+                    .col(ColumnDef::new(Alias::new("published_at")).timestamp_with_time_zone().null())
+                    .col(ColumnDef::new(Alias::new("archived_at")).timestamp_with_time_zone().not_null().default(Expr::current_timestamp()))
+                    .to_owned(),
+            )
+            .await?;
+
+        manager
+            .create_index(
+                Index::create()
+                    .name("idx_domain_events_archive_created")
+                    .table(Alias::new("domain_events_archive"))
+                    .col(Alias::new("created_at"))
+                    .to_owned(),
+            )
+            .await?;
+
+        // 清理函数：将 >90 天的已发布事件迁移到归档表
+        manager
+            .get_connection()
+            .execute_unprepared(
+                r#"
+                CREATE OR REPLACE FUNCTION cleanup_old_published_events(
+                    retention_days INT DEFAULT 90,
+                    batch_size INT DEFAULT 1000
+                ) RETURNS INT AS $$
+                DECLARE
+                    moved_count INT;
+                BEGIN
+                    INSERT INTO domain_events_archive (id, tenant_id, event_type, payload, correlation_id, status, attempts, last_error, created_at, published_at)
+                    SELECT id, tenant_id, event_type, payload, correlation_id, status, attempts, last_error, created_at, published_at
+                    FROM domain_events
+                    WHERE status = 'published'
+                      AND published_at < NOW() - (retention_days || ' days')::INTERVAL
+                    ORDER BY created_at ASC
+                    LIMIT batch_size;
+
+                    GET DIAGNOSTICS moved_count = ROW_COUNT;
+
+                    DELETE FROM domain_events
+                    WHERE status = 'published'
+                      AND published_at < NOW() - (retention_days || ' days')::INTERVAL
+                    LIMIT batch_size;
+
+                    RETURN moved_count;
+                END;
+                $$ LANGUAGE plpgsql;
+                "#,
+            )
+            .await?;
+
+        Ok(())
+    }
+
+    async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+        manager
+            .get_connection()
+            .execute_unprepared("DROP FUNCTION IF EXISTS cleanup_old_published_events(INT, INT);")
+            .await?;
+
+        manager
+            .drop_table(Table::drop().table(Alias::new("domain_events_archive")).to_owned())
+            .await?;
+
+        Ok(())
+    }
+}

crates/erp-server/migration/src/m20260427_000085_processed_events.rs

@@ -0,0 +1,61 @@
+use sea_orm_migration::prelude::*;
+
+#[derive(DeriveMigrationName)]
+pub struct Migration;
+
+#[async_trait::async_trait]
+impl MigrationTrait for Migration {
+    async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+        manager
+            .create_table(
+                Table::create()
+                    .table(Alias::new("processed_events"))
+                    .if_not_exists()
+                    .col(ColumnDef::new(Alias::new("event_id")).uuid().not_null())
+                    .col(ColumnDef::new(Alias::new("consumer_id")).string_len(200).not_null())
+                    .col(ColumnDef::new(Alias::new("processed_at")).timestamp_with_time_zone().not_null().default(Expr::current_timestamp()))
+                    .primary_key(Index::create().col(Alias::new("event_id")).col(Alias::new("consumer_id")))
+                    .to_owned(),
+            )
+            .await?;
+
+        // 7 天 TTL 清理函数
+        manager
+            .get_connection()
+            .execute_unprepared(
+                r#"
+                CREATE OR REPLACE FUNCTION cleanup_old_processed_events(
+                    retention_days INT DEFAULT 7,
+                    batch_size INT DEFAULT 1000
+                ) RETURNS INT AS $$
+                DECLARE
+                    deleted_count INT;
+                BEGIN
+                    DELETE FROM processed_events
+                    WHERE processed_at < NOW() - (retention_days || ' days')::INTERVAL
+                    LIMIT batch_size;
+
+                    GET DIAGNOSTICS deleted_count = ROW_COUNT;
+                    RETURN deleted_count;
+                END;
+                $$ LANGUAGE plpgsql;
+                "#,
+            )
+            .await?;
+
+        Ok(())
+    }
+
+    async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+        manager
+            .get_connection()
+            .execute_unprepared("DROP FUNCTION IF EXISTS cleanup_old_processed_events(INT, INT);")
+            .await?;
+
+        manager
+            .drop_table(Table::drop().table(Alias::new("processed_events")).to_owned())
+            .await?;
+
+        Ok(())
+    }
+}

crates/erp-server/src/main.rs

@@ -4,6 +4,7 @@ mod handlers;
 mod middleware;
 mod outbox;
 mod state;
+mod tasks;
 
 /// OpenAPI 规范定义 — 通过 utoipa derive 合并各模块 schema。
 #[derive(OpenApi)]
@@ -406,10 +407,13 @@ async fn main() -> anyhow::Result<()> {
     erp_plugin::notification::start_notification_listener(db.clone(), event_bus.clone());
     tracing::info!("Plugin notification listener started");
 
-    // Start outbox relay (re-publish pending domain events)
-    outbox::start_outbox_relay(db.clone(), event_bus.clone());
+    // Start outbox relay (LISTEN/NOTIFY + fallback poll for pending domain events)
+    outbox::start_outbox_relay(db.clone(), event_bus.clone(), config.database.url.clone());
     tracing::info!("Outbox relay started");
 
+    // Start event cleanup (archive old published events + purge processed_events)
+    tasks::start_event_cleanup(db.clone());
+
     // Start timeout checker (scan overdue tasks every 60s)
     erp_workflow::WorkflowModule::start_timeout_checker(db.clone(), event_bus.clone());
     tracing::info!("Timeout checker started");

crates/erp-server/src/outbox.rs

@@ -1,19 +1,29 @@
 use chrono::Utc;
 use sea_orm::{ActiveModelTrait, ColumnTrait, EntityTrait, QueryFilter, QueryOrder, QuerySelect, Set};
+use sqlx::postgres::PgListener;
 use std::time::Duration;
 
 use erp_core::entity::domain_event;
 use erp_core::events::{DomainEvent, EventBus};
 
 const MAX_RETRY: i32 = 5;
+const FALLBACK_POLL_INTERVAL_SECS: u64 = 30;
+const NOTIFY_CHANNEL: &str = "outbox_channel";
+const RECONNECT_DELAY_SECS: u64 = 5;
 
 /// 启动 outbox relay 后台任务。
 ///
 /// 先执行一次性扫描（处理服务重启前遗留的 pending 事件），
-/// 然后每 5 秒定期扫描 domain_events 表中 status = 'pending' 的事件。
-pub fn start_outbox_relay(db: sea_orm::DatabaseConnection, event_bus: EventBus) {
+/// 然后通过 PostgreSQL LISTEN/NOTIFY 监听新事件，配合 30s 兜底轮询。
+pub fn start_outbox_relay(
+    db: sea_orm::DatabaseConnection,
+    event_bus: EventBus,
+    database_url: String,
+) {
     let db_clone = db.clone();
     let event_bus_clone = event_bus.clone();
+    let url = database_url.clone();
+
     tokio::spawn(async move {
         // 启动时立即处理一次（恢复重启前未广播的事件）
         match process_pending_events(&db_clone, &event_bus_clone).await {
@@ -22,17 +32,65 @@ pub fn start_outbox_relay(db: sea_orm::DatabaseConnection, event_bus: EventBus)
             Err(e) => tracing::warn!(error = %e, "启动时 outbox relay 处理失败"),
         }
 
-        // 定期轮询
-        let mut interval = tokio::time::interval(Duration::from_secs(5));
+        // 进入 LISTEN/NOTIFY 主循环（带自动重连）
         loop {
-            interval.tick().await;
+            if let Err(e) = run_listener(&db_clone, &event_bus_clone, &url).await {
+                tracing::warn!(error = %e, "PgListener 断开连接，{}s 后重连", RECONNECT_DELAY_SECS);
+            }
+            tokio::time::sleep(Duration::from_secs(RECONNECT_DELAY_SECS)).await;
+
+            // 重连后执行一次兜底扫描
             if let Err(e) = process_pending_events(&db_clone, &event_bus_clone).await {
-                tracing::warn!(error = %e, "Outbox relay 处理失败");
+                tracing::warn!(error = %e, "重连后 outbox relay 处理失败");
             }
         }
     });
 }
 
+/// 运行 PgListener 监听循环。
+///
+/// 使用 `tokio::select!` 在 LISTEN 通知和 30s 定时器之间竞争，
+/// 确保即使 NOTIFY 丢失也能兜底处理。
+async fn run_listener(
+    db: &sea_orm::DatabaseConnection,
+    event_bus: &EventBus,
+    database_url: &str,
+) -> Result<(), sqlx::Error> {
+    let mut listener = PgListener::connect(database_url).await?;
+    listener.listen(NOTIFY_CHANNEL).await?;
+    tracing::info!("Outbox relay LISTEN/NOTIFY 已连接，监听 {}", NOTIFY_CHANNEL);
+
+    let mut fallback = tokio::time::interval(Duration::from_secs(FALLBACK_POLL_INTERVAL_SECS));
+
+    loop {
+        tokio::select! {
+            // LISTEN/NOTIFY 通知触发
+            notification = listener.recv() => {
+                match notification {
+                    Ok(notif) => {
+                        tracing::debug!(
+                            channel = %notif.channel(),
+                            payload = %notif.payload(),
+                            "收到 outbox NOTIFY"
+                        );
+                        if let Err(e) = process_pending_events(db, event_bus).await {
+                            tracing::warn!(error = %e, "NOTIFY 触发的 outbox 处理失败");
+                        }
+                    }
+                    Err(e) => return Err(e),
+                }
+            }
+            // 30s 兜底轮询
+            _ = fallback.tick() => {
+                tracing::debug!("outbox relay 兜底轮询触发");
+                if let Err(e) = process_pending_events(db, event_bus).await {
+                    tracing::warn!(error = %e, "兜底轮询 outbox 处理失败");
+                }
+            }
+        }
+    }
+}
+
 async fn process_pending_events(
     db: &sea_orm::DatabaseConnection,
     event_bus: &EventBus,

crates/erp-server/src/tasks.rs

@@ -0,0 +1,53 @@
+use std::time::Duration;
+
+/// 启动事件清理后台任务。
+///
+/// 每日执行一次：
+/// - 调用 `cleanup_old_published_events()` 归档 >90 天的已发布事件
+/// - 调用 `cleanup_old_processed_events()` 清理 >7 天的去重记录
+pub fn start_event_cleanup(db: sea_orm::DatabaseConnection) {
+    tokio::spawn(async move {
+        let mut interval = tokio::time::interval(Duration::from_secs(86400));
+        loop {
+            interval.tick().await;
+            if let Err(e) = run_cleanup(&db).await {
+                tracing::warn!(error = %e, "事件清理任务执行失败");
+            }
+        }
+    });
+    tracing::info!("事件清理任务已启动（每 24 小时执行一次）");
+}
+
+async fn run_cleanup(db: &sea_orm::DatabaseConnection) -> Result<(), sea_orm::DbErr> {
+    use sea_orm::ConnectionTrait;
+
+    // 归档 >90 天的已发布事件
+    match db
+        .execute_unprepared("SELECT cleanup_old_published_events(90, 1000)")
+        .await
+    {
+        Ok(result) => {
+            tracing::info!(
+                rows_affected = result.rows_affected(),
+                "已发布事件归档完成"
+            );
+        }
+        Err(e) => tracing::warn!(error = %e, "已发布事件归档失败"),
+    }
+
+    // 清理 >7 天的去重记录
+    match db
+        .execute_unprepared("SELECT cleanup_old_processed_events(7, 1000)")
+        .await
+    {
+        Ok(result) => {
+            tracing::info!(
+                rows_affected = result.rows_affected(),
+                "去重记录清理完成"
+            );
+        }
+        Err(e) => tracing::warn!(error = %e, "去重记录清理失败"),
+    }
+
+    Ok(())
+}

crates/erp-workflow/src/module.rs

@@ -115,11 +115,11 @@ impl WorkflowModule {
                                 let event = erp_core::events::DomainEvent::new(
                                     "task.timeout",
                                     *tenant_id,
-                                    serde_json::json!({
+                                    erp_core::events::build_event_payload(serde_json::json!({
                                         "task_id": task_id,
                                         "instance_id": instance_id,
                                         "assignee_id": assignee_id,
-                                    }),
+                                    })),
                                 );
                                 event_bus.publish(event, &db).await;
                             }

crates/erp-workflow/src/service/instance_service.rs

@@ -126,7 +126,7 @@ impl InstanceService {
         event_bus.publish(erp_core::events::DomainEvent::new(
             "process_instance.started",
             tenant_id,
-            serde_json::json!({ "instance_id": instance_id, "definition_id": definition.id, "started_by": operator_id }),
+            erp_core::events::build_event_payload(serde_json::json!({ "instance_id": instance_id, "definition_id": definition.id, "started_by": operator_id })),
         ), db).await;
 
         audit_service::record(
@@ -323,7 +323,7 @@ impl InstanceService {
             tenant_id: Set(tenant_id),
             event_type: Set(event_type),
             payload: Set(Some(
-                serde_json::json!({ "instance_id": id, "changed_by": operator_id }),
+                erp_core::events::build_event_payload(serde_json::json!({ "instance_id": id, "changed_by": operator_id })),
             )),
             correlation_id: Set(Some(Uuid::now_v7())),
             status: Set("pending".to_string()),