d5c9654370
- 084/085: PostgreSQL DELETE 不支持 LIMIT，改用 ctid IN (SELECT ... LIMIT) - 086: RLS 迁移改为动态查询 information_schema 获取含 tenant_id 的表， 避免硬编码表名不一致问题 - 全量测试 490 个通过（含 27 个集成测试 + RLS 验证）
94 lines
3.9 KiB
Rust
94 lines
3.9 KiB
Rust
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 ctid IN (
|
|
SELECT ctid FROM domain_events
|
|
WHERE status = 'published'
|
|
AND published_at < NOW() - (retention_days || ' days')::INTERVAL
|
|
ORDER BY created_at ASC
|
|
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(())
|
|
}
|
|
}
|