feat(core): EventBus dead-letter + consume_with_retry 辅助函数
Some checks failed
CI / rust-check (push) Has been cancelled
CI / rust-test (push) Has been cancelled
CI / frontend-build (push) Has been cancelled
CI / security-audit (push) Has been cancelled

- 新增 dead_letter_events 表 + Entity
- consume_with_retry: 幂等检查 + 成功标记 + 失败转入 dead-letter
- insert_dead_letter: 写入失败事件供后续排查和手动重试
This commit is contained in:
iven
2026-04-28 11:47:44 +08:00
parent 10755cde0e
commit be8fca1d76
5 changed files with 192 additions and 0 deletions

View File

@@ -90,6 +90,7 @@ mod m20260427_000087_audit_logs_hash_chain;
mod m20260428_000088_rls_policy_strict;
mod m20260428_000089_blind_indexes;
mod m20260428_000090_critical_alerts;
mod m20260428_000091_dead_letter_events;
pub struct Migrator;
@@ -187,6 +188,7 @@ impl MigratorTrait for Migrator {
Box::new(m20260428_000088_rls_policy_strict::Migration),
Box::new(m20260428_000089_blind_indexes::Migration),
Box::new(m20260428_000090_critical_alerts::Migration),
Box::new(m20260428_000091_dead_letter_events::Migration),
]
}
}

View File

@@ -0,0 +1,76 @@
use sea_orm_migration::prelude::*;
#[derive(DeriveMigrationName)]
pub struct Migration;
#[derive(Iden)]
enum DeadLetterEvent {
Table,
Id,
TenantId,
OriginalEventId,
EventType,
Payload,
ConsumerId,
Attempts,
LastError,
CreatedAt,
ResolvedAt,
}
#[async_trait::async_trait]
impl MigrationTrait for Migration {
async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {
manager
.create_table(
Table::create()
.table(DeadLetterEvent::Table)
.col(
ColumnDef::new(DeadLetterEvent::Id)
.uuid()
.not_null()
.primary_key()
.default(PgFunc::gen_random_uuid()),
)
.col(ColumnDef::new(DeadLetterEvent::TenantId).uuid())
.col(
ColumnDef::new(DeadLetterEvent::OriginalEventId)
.uuid()
.not_null(),
)
.col(
ColumnDef::new(DeadLetterEvent::EventType)
.string_len(128)
.not_null(),
)
.col(ColumnDef::new(DeadLetterEvent::Payload).json_binary())
.col(
ColumnDef::new(DeadLetterEvent::ConsumerId)
.string_len(128)
.not_null(),
)
.col(
ColumnDef::new(DeadLetterEvent::Attempts)
.integer()
.not_null()
.default(0),
)
.col(ColumnDef::new(DeadLetterEvent::LastError).text())
.col(
ColumnDef::new(DeadLetterEvent::CreatedAt)
.timestamp_with_time_zone()
.not_null()
.default(Expr::current_timestamp()),
)
.col(ColumnDef::new(DeadLetterEvent::ResolvedAt).timestamp_with_time_zone())
.to_owned(),
)
.await
}
async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> {
manager
.drop_table(Table::drop().table(DeadLetterEvent::Table).to_owned())
.await
}
}