feat(plugin): Keyset Pagination — cursor 编解码 + 游标分页 SQL
This commit is contained in:
@@ -22,3 +22,4 @@ axum = { workspace = true }
|
||||
utoipa = { workspace = true }
|
||||
async-trait = { workspace = true }
|
||||
sha2 = { workspace = true }
|
||||
base64 = "0.22"
|
||||
|
||||
@@ -29,6 +29,8 @@ pub struct UpdatePluginDataReq {
|
||||
pub struct PluginDataListParams {
|
||||
pub page: Option<u64>,
|
||||
pub page_size: Option<u64>,
|
||||
/// Base64 编码的游标(用于 Keyset 分页)
|
||||
pub cursor: Option<String>,
|
||||
pub search: Option<String>,
|
||||
/// JSON 格式过滤: {"field":"value"}
|
||||
pub filter: Option<String>,
|
||||
|
||||
@@ -1,6 +1,8 @@
|
||||
use sea_orm::{ConnectionTrait, DatabaseConnection, FromQueryResult, Statement, Value};
|
||||
use uuid::Uuid;
|
||||
|
||||
use base64::{Engine, engine::general_purpose::STANDARD as BASE64};
|
||||
|
||||
use crate::error::{PluginError, PluginResult};
|
||||
use crate::manifest::{PluginEntity, PluginFieldType};
|
||||
|
||||
@@ -597,6 +599,90 @@ impl DynamicTableManager {
|
||||
|
||||
Ok((sql, values))
|
||||
}
|
||||
|
||||
/// 编码游标
|
||||
pub fn encode_cursor(values: &[String], id: &Uuid) -> String {
|
||||
let obj = serde_json::json!({
|
||||
"v": values,
|
||||
"id": id.to_string(),
|
||||
});
|
||||
BASE64.encode(obj.to_string())
|
||||
}
|
||||
|
||||
/// 解码游标
|
||||
pub fn decode_cursor(cursor: &str) -> Result<(Vec<String>, Uuid), String> {
|
||||
let json_str = BASE64
|
||||
.decode(cursor)
|
||||
.map_err(|e| format!("游标 Base64 解码失败: {}", e))?;
|
||||
let obj: serde_json::Value = serde_json::from_slice(&json_str)
|
||||
.map_err(|e| format!("游标 JSON 解析失败: {}", e))?;
|
||||
let values = obj["v"]
|
||||
.as_array()
|
||||
.ok_or("游标缺少 v 字段")?
|
||||
.iter()
|
||||
.map(|v| v.as_str().unwrap_or("").to_string())
|
||||
.collect();
|
||||
let id = obj["id"].as_str().ok_or("游标缺少 id 字段")?;
|
||||
let id = Uuid::parse_str(id).map_err(|e| format!("游标 id 解析失败: {}", e))?;
|
||||
Ok((values, id))
|
||||
}
|
||||
|
||||
/// 构建 Keyset 分页 SQL
|
||||
pub fn build_keyset_query_sql(
|
||||
table_name: &str,
|
||||
tenant_id: Uuid,
|
||||
limit: u64,
|
||||
cursor: Option<String>,
|
||||
sort_column: Option<String>,
|
||||
sort_direction: &str,
|
||||
generated_fields: &[String],
|
||||
) -> Result<(String, Vec<Value>), String> {
|
||||
let dir = match sort_direction {
|
||||
"ASC" => "ASC",
|
||||
_ => "DESC",
|
||||
};
|
||||
let ref_fn = Self::field_reference_fn(generated_fields);
|
||||
let sort_col = sort_column
|
||||
.as_deref()
|
||||
.map(|s| ref_fn(s))
|
||||
.unwrap_or("\"created_at\"".to_string());
|
||||
|
||||
let mut values: Vec<Value> = vec![tenant_id.into()];
|
||||
let mut param_idx = 2;
|
||||
|
||||
let cursor_condition = if let Some(c) = cursor {
|
||||
let (sort_vals, cursor_id) = Self::decode_cursor(&c)?;
|
||||
let cond = format!(
|
||||
"ROW({}, \"id\") {} (${}, ${})",
|
||||
sort_col,
|
||||
if dir == "ASC" { ">" } else { "<" },
|
||||
param_idx,
|
||||
param_idx + 1
|
||||
);
|
||||
values.push(Value::String(Some(Box::new(
|
||||
sort_vals.first().cloned().unwrap_or_default(),
|
||||
))));
|
||||
values.push(cursor_id.into());
|
||||
param_idx += 2;
|
||||
Some(cond)
|
||||
} else {
|
||||
None
|
||||
};
|
||||
|
||||
let where_extra = cursor_condition
|
||||
.map(|c| format!(" AND {}", c))
|
||||
.unwrap_or_default();
|
||||
|
||||
let sql = format!(
|
||||
"SELECT id, data, created_at, updated_at, version FROM \"{}\" \
|
||||
WHERE \"tenant_id\" = $1 AND \"deleted_at\" IS NULL{} \
|
||||
ORDER BY {}, \"id\" {} LIMIT ${}",
|
||||
table_name, where_extra, sort_col, dir, param_idx,
|
||||
);
|
||||
values.push((limit as i64).into());
|
||||
|
||||
Ok((sql, values))
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
@@ -978,4 +1064,62 @@ mod tests {
|
||||
sql
|
||||
);
|
||||
}
|
||||
|
||||
// ===== Keyset Pagination 测试 =====
|
||||
|
||||
#[test]
|
||||
fn test_keyset_cursor_encode_decode() {
|
||||
let cursor = DynamicTableManager::encode_cursor(
|
||||
&["测试值".to_string()],
|
||||
&Uuid::parse_str("00000000-0000-0000-0000-000000000042").unwrap(),
|
||||
);
|
||||
let decoded = DynamicTableManager::decode_cursor(&cursor).unwrap();
|
||||
assert_eq!(decoded.0, vec!["测试值"]);
|
||||
assert_eq!(
|
||||
decoded.1,
|
||||
Uuid::parse_str("00000000-0000-0000-0000-000000000042").unwrap()
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_keyset_sql_first_page() {
|
||||
let (sql, _) = DynamicTableManager::build_keyset_query_sql(
|
||||
"plugin_test",
|
||||
Uuid::parse_str("00000000-0000-0000-0000-000000000001").unwrap(),
|
||||
20,
|
||||
None,
|
||||
Some("_f_name".to_string()),
|
||||
"ASC",
|
||||
&[],
|
||||
)
|
||||
.unwrap();
|
||||
assert!(sql.contains("ORDER BY"), "应有 ORDER BY");
|
||||
assert!(!sql.contains("ROW("), "第一页不应有 cursor 条件");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_keyset_sql_with_cursor() {
|
||||
let cursor = DynamicTableManager::encode_cursor(
|
||||
&["Alice".to_string()],
|
||||
&Uuid::parse_str("00000000-0000-0000-0000-000000000100").unwrap(),
|
||||
);
|
||||
let (sql, values) = DynamicTableManager::build_keyset_query_sql(
|
||||
"plugin_test",
|
||||
Uuid::parse_str("00000000-0000-0000-0000-000000000001").unwrap(),
|
||||
20,
|
||||
Some(cursor),
|
||||
Some("_f_name".to_string()),
|
||||
"ASC",
|
||||
&[],
|
||||
)
|
||||
.unwrap();
|
||||
assert!(
|
||||
sql.contains("ROW("),
|
||||
"cursor 条件应使用 ROW 比较"
|
||||
);
|
||||
assert!(
|
||||
values.len() >= 4,
|
||||
"应有 tenant_id + cursor_val + cursor_id + limit"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user