d5c9654370
- 084/085: PostgreSQL DELETE 不支持 LIMIT，改用 ctid IN (SELECT ... LIMIT) - 086: RLS 迁移改为动态查询 information_schema 获取含 tenant_id 的表， 避免硬编码表名不一致问题 - 全量测试 490 个通过（含 27 个集成测试 + RLS 验证）
65 lines
2.2 KiB
Rust
65 lines
2.2 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> {
|
|
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 ctid IN (
|
|
SELECT ctid 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(())
|
|
}
|
|
}
|