GenarrativeAI/Genarrative
5
0
Fork 0
Code Issues Pull Requests Actions 5 Packages Projects Releases Wiki Activity

feat(admin): add database table query page

Browse Source
This commit is contained in:
历冰郁-hermes版
2026-05-08 16:39:44 +08:00
parent b08127031c
commit 72fce47187
12 changed files with 964 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

424
server-rs/crates/api-server/src/admin.rs
View File

@@ -16,12 +16,14 @@ use axum::{
};
use reqwest::Client;
use serde::Deserialize;
use serde_json::Value;
use serde_json::{Map, Value};
use shared_contracts::admin::{
AdminDatabaseOverviewPayload, AdminDatabaseTableStatPayload, AdminDebugHeaderInput,
AdminDebugHttpRequest, AdminDebugHttpResponse, AdminLoginRequest, AdminLoginResponse,
AdminMeResponse, AdminOverviewResponse, AdminServiceOverviewPayload, AdminSessionPayload,
AdminTrackingEventEntryPayload, AdminTrackingEventListQuery, AdminTrackingEventListResponse,
AdminDatabaseOverviewPayload, AdminDatabaseTableListResponse, AdminDatabaseTableRowPayload,
AdminDatabaseTableRowsQuery, AdminDatabaseTableRowsResponse, AdminDatabaseTableStatPayload,
AdminDebugHeaderInput, AdminDebugHttpRequest, AdminDebugHttpResponse, AdminLoginRequest,
AdminLoginResponse, AdminMeResponse, AdminOverviewResponse, AdminServiceOverviewPayload,
AdminSessionPayload, AdminTrackingEventEntryPayload, AdminTrackingEventListQuery,
AdminTrackingEventListResponse,
};
use time::{OffsetDateTime, format_description::well_known::Rfc3339};
@@ -46,6 +48,8 @@ const BLOCKED_DEBUG_HEADERS: &[&str] = &[
const SPACETIME_SCHEMA_VERSION_QUERY: &str = "version=9";
const ADMIN_TRACKING_EVENT_DEFAULT_LIMIT: u32 = 200;
const ADMIN_TRACKING_EVENT_MAX_LIMIT: u32 = 1000;
const ADMIN_DATABASE_TABLE_DEFAULT_LIMIT: u32 = 100;
const ADMIN_DATABASE_TABLE_MAX_LIMIT: u32 = 500;
#[derive(Clone, Debug)]
pub struct AuthenticatedAdmin {
@@ -170,6 +174,26 @@ pub async fn admin_list_tracking_events(
))
}
pub async fn admin_list_database_tables(
State(state): State<AppState>,
Extension(request_context): Extension<RequestContext>,
Extension(_admin): Extension<AuthenticatedAdmin>,
) -> Result<Json<Value>, AppError> {
let response = fetch_admin_database_table_list(&state).await?;
Ok(json_success_body(Some(&request_context), response))
}
pub async fn admin_list_database_table_rows(
State(state): State<AppState>,
Extension(request_context): Extension<RequestContext>,
Extension(_admin): Extension<AuthenticatedAdmin>,
axum::extract::Path(table_name): axum::extract::Path<String>,
Query(query): Query<AdminDatabaseTableRowsQuery>,
) -> Result<Json<Value>, AppError> {
let response = fetch_admin_database_table_rows(&state, &table_name, query).await?;
Ok(json_success_body(Some(&request_context), response))
}
pub async fn require_admin_auth(
State(state): State<AppState>,
mut request: Request,
@@ -263,21 +287,7 @@ async fn fetch_database_overview(state: &AppState) -> AdminDatabaseOverviewPaylo
.ok()
.flatten();
let schema_table_names = schema
.as_ref()
.and_then(|value| value.tables.as_ref())
.map(|tables| {
tables
.iter()
.filter_map(|table| table.name.as_deref())
.map(str::trim)
.filter(|name| !name.is_empty())
.map(ToOwned::to_owned)
.collect::<BTreeSet<_>>()
.into_iter()
.collect::<Vec<_>>()
})
.unwrap_or_default();
let schema_table_names = extract_schema_table_names(schema.as_ref());
let mut table_stats = Vec::new();
for table_name in &schema_table_names {
@@ -505,6 +515,275 @@ fn parse_count_value(value: &Value) -> Result<u64, String> {
}
}
async fn fetch_admin_database_table_list(
state: &AppState,
) -> Result<AdminDatabaseTableListResponse, AppError> {
let (_, tables, fetch_errors) = fetch_admin_database_schema_tables(state).await;
Ok(AdminDatabaseTableListResponse {
tables,
fetch_errors,
})
}
async fn fetch_admin_database_table_rows(
state: &AppState,
table_name: &str,
query: AdminDatabaseTableRowsQuery,
) -> Result<AdminDatabaseTableRowsResponse, AppError> {
let table_name = table_name.trim();
if !is_safe_spacetime_table_name(table_name) {
return Err(AppError::from_status(StatusCode::BAD_REQUEST).with_message("表名不合法"));
}
let (_, tables, _) = fetch_admin_database_schema_tables(state).await;
if !tables.iter().any(|name| name == table_name) {
return Err(AppError::from_status(StatusCode::NOT_FOUND).with_message("表不存在"));
}
let client = Client::new();
let server_root = state.config.spacetime_server_url.trim_end_matches('/');
let database = state.config.spacetime_database.trim();
let token = resolve_admin_spacetime_sql_token(state);
let limit = clamp_admin_database_table_limit(query.limit);
let sql = format!("SELECT * FROM {table_name} LIMIT {limit}");
let payload = fetch_spacetime_sql_json(&client, server_root, database, token.as_deref(), &sql)
.await
.map_err(|error| {
AppError::from_status(StatusCode::BAD_GATEWAY).with_message(format!(
"表数据读取失败：{}",
normalize_table_count_error(&error)
))
})?;
let mut response = parse_admin_database_table_rows_sql_response(table_name, limit, payload)
.map_err(|error| {
AppError::from_status(StatusCode::BAD_GATEWAY)
.with_message(format!("表数据解析失败：{error}"))
})?;
apply_admin_database_table_filters(&mut response.rows, &query)?;
response.total_returned = response.rows.len();
Ok(response)
}
async fn fetch_admin_database_schema_tables(
state: &AppState,
) -> (Option<SpacetimeSchemaResponse>, Vec<String>, Vec<String>) {
let client = Client::new();
let server_root = state.config.spacetime_server_url.trim_end_matches('/');
let database = state.config.spacetime_database.trim();
let token = resolve_admin_spacetime_sql_token(state);
let mut fetch_errors = Vec::new();
let schema = fetch_spacetime_json::<SpacetimeSchemaResponse>(
&client,
&build_spacetime_schema_url(server_root, database),
token.as_deref(),
)
.await
.map_err(|error| fetch_errors.push(format!("数据库 schema 读取失败：{error}")))
.ok()
.flatten();
let tables = extract_schema_table_names(schema.as_ref());
(schema, tables, fetch_errors)
}
fn extract_schema_table_names(schema: Option<&SpacetimeSchemaResponse>) -> Vec<String> {
schema
.and_then(|value| value.tables.as_ref())
.map(|tables| {
tables
.iter()
.filter_map(|table| table.name.as_deref())
.map(str::trim)
.filter(|name| !name.is_empty())
.map(ToOwned::to_owned)
.collect::<BTreeSet<_>>()
.into_iter()
.collect::<Vec<_>>()
})
.unwrap_or_default()
}
fn resolve_admin_spacetime_sql_token(state: &AppState) -> Option<String> {
state
.config
.spacetime_token
.as_deref()
.map(str::trim)
.filter(|value| !value.is_empty())
.map(str::to_string)
.or_else(load_local_spacetime_cli_token)
}
fn clamp_admin_database_table_limit(limit: Option<u32>) -> u32 {
limit
.unwrap_or(ADMIN_DATABASE_TABLE_DEFAULT_LIMIT)
.clamp(1, ADMIN_DATABASE_TABLE_MAX_LIMIT)
}
fn parse_admin_database_table_rows_sql_response(
table_name: &str,
limit: u32,
payload: Value,
) -> Result<AdminDatabaseTableRowsResponse, String> {
let statement = extract_first_sql_statement(payload)?;
let columns = extract_sql_statement_columns(&statement);
let rows_value = statement
.get("rows")
.ok_or_else(|| "SQL 响应缺少 rows 字段".to_string())?;
let row_values = rows_value
.as_array()
.ok_or_else(|| "SQL rows 字段格式非法".to_string())?;
let rows = row_values
.iter()
.map(|row| build_admin_database_table_row(row, &columns))
.collect::<Vec<_>>();
Ok(AdminDatabaseTableRowsResponse {
table_name: table_name.to_string(),
columns,
total_returned: rows.len(),
rows,
limit,
})
}
fn extract_first_sql_statement(payload: Value) -> Result<Value, String> {
match payload {
Value::Array(statements) => statements
.into_iter()
.next()
.ok_or_else(|| "SQL 结果为空".to_string()),
Value::Object(statement) => Ok(Value::Object(statement)),
_ => Err("SQL 响应格式非法".to_string()),
}
}
fn extract_sql_statement_columns(statement: &Value) -> Vec<String> {
statement
.get("schema")
.and_then(|schema| schema.get("elements"))
.and_then(Value::as_array)
.map(|elements| {
elements
.iter()
.enumerate()
.map(|(index, element)| {
element
.get("name")
.and_then(extract_sql_schema_name)
.map(ToOwned::to_owned)
.unwrap_or_else(|| format!("col_{}", index + 1))
})
.collect::<Vec<_>>()
})
.unwrap_or_default()
}
fn build_admin_database_table_row(row: &Value, columns: &[String]) -> AdminDatabaseTableRowPayload {
let raw = normalize_admin_database_value(row);
let mut cells = Map::new();
if let Some(values) = row.as_array() {
for (index, value) in values.iter().enumerate() {
let key = columns
.get(index)
.cloned()
.unwrap_or_else(|| format!("col_{}", index + 1));
cells.insert(key, normalize_admin_database_value(value));
}
} else if let Some(object) = row.as_object() {
for (key, value) in object {
cells.insert(key.clone(), normalize_admin_database_value(value));
}
}
AdminDatabaseTableRowPayload {
cells: Value::Object(cells),
raw,
}
}
fn normalize_admin_database_value(value: &Value) -> Value {
match value {
Value::Array(items) if items.len() == 1 => normalize_admin_database_value(&items[0]),
Value::Array(items) if items.len() == 2 => {
if let Some(index) = items.first().and_then(Value::as_u64) {
if index == 0 {
return items
.get(1)
.map(normalize_admin_database_value)
.unwrap_or(Value::Null);
}
if index == 1 && items.get(1).and_then(Value::as_array).is_some() {
return Value::Null;
}
}
Value::Array(items.iter().map(normalize_admin_database_value).collect())
}
Value::Array(items) => {
Value::Array(items.iter().map(normalize_admin_database_value).collect())
}
Value::Object(object) => {
if let Some(value) = object.get("some") {
return normalize_admin_database_value(value);
}
Value::Object(
object
.iter()
.map(|(key, value)| (key.clone(), normalize_admin_database_value(value)))
.collect(),
)
}
_ => value.clone(),
}
}
fn apply_admin_database_table_filters(
rows: &mut Vec<AdminDatabaseTableRowPayload>,
query: &AdminDatabaseTableRowsQuery,
) -> Result<(), AppError> {
if let Some(search) = normalized_non_empty(query.search.as_deref()) {
let needle = search.to_ascii_lowercase();
rows.retain(|row| row.cells.to_string().to_ascii_lowercase().contains(&needle));
}
if let Some(filters) = normalized_non_empty(query.filters.as_deref()) {
let parsed = serde_json::from_str::<Value>(filters).map_err(|error| {
AppError::from_status(StatusCode::BAD_REQUEST)
.with_message(format!("筛选 JSON 解析失败：{error}"))
})?;
let object = parsed.as_object().ok_or_else(|| {
AppError::from_status(StatusCode::BAD_REQUEST)
.with_message("筛选条件必须是 JSON object")
})?;
rows.retain(|row| row_matches_admin_database_filters(row, object));
}
Ok(())
}
fn row_matches_admin_database_filters(
row: &AdminDatabaseTableRowPayload,
filters: &Map<String, Value>,
) -> bool {
let Some(cells) = row.cells.as_object() else {
return filters.is_empty();
};
filters.iter().all(|(key, expected)| {
cells
.get(key)
.map(|actual| admin_database_filter_value_matches(actual, expected))
.unwrap_or(false)
})
}
fn admin_database_filter_value_matches(actual: &Value, expected: &Value) -> bool {
if actual == expected {
return true;
}
if let Some(expected_text) = expected.as_str() {
return value_to_string(actual)
.map(|actual_text| actual_text == expected_text)
.unwrap_or(false);
}
false
}
async fn fetch_admin_tracking_events(
state: &AppState,
query: AdminTrackingEventListQuery,
@@ -949,14 +1228,16 @@ fn build_admin_session_payload(session: crate::state::AdminSession) -> AdminSess
#[cfg(test)]
mod tests {
use super::{
apply_admin_database_table_filters, build_admin_database_table_row,
build_admin_tracking_events_sql, build_body_preview, build_debug_base_url,
build_spacetime_schema_url, clamp_admin_tracking_event_limit, is_safe_spacetime_table_name,
normalize_debug_path, normalize_table_count_error,
build_spacetime_schema_url, clamp_admin_database_table_limit,
clamp_admin_tracking_event_limit, is_safe_spacetime_table_name, normalize_debug_path,
normalize_table_count_error, parse_admin_database_table_rows_sql_response,
parse_admin_tracking_events_sql_response, parse_spacetime_sql_count_response, trim_preview,
};
use axum::{http::StatusCode, response::IntoResponse};
use serde_json::json;
use shared_contracts::admin::AdminTrackingEventListQuery;
use shared_contracts::admin::{AdminDatabaseTableRowsQuery, AdminTrackingEventListQuery};
#[test]
fn normalize_debug_path_rejects_absolute_url() {
@@ -1119,6 +1400,103 @@ mod tests {
assert_eq!(count, 3);
}
#[test]
fn clamp_admin_database_table_limit_uses_default_and_bounds() {
assert_eq!(clamp_admin_database_table_limit(None), 100);
assert_eq!(clamp_admin_database_table_limit(Some(0)), 1);
assert_eq!(clamp_admin_database_table_limit(Some(800)), 500);
}
#[test]
fn parse_admin_database_table_rows_sql_response_maps_schema_columns() {
let payload = json!([
{
"schema": {
"elements": [
{"name": {"some": "user_id"}},
{"name": {"some": "points"}}
]
},
"rows": [["u1", 12]]
}
]);
let response = parse_admin_database_table_rows_sql_response("profile_wallet", 100, payload)
.expect("table rows should parse");
assert_eq!(response.table_name, "profile_wallet");
assert_eq!(response.columns, vec!["user_id", "points"]);
assert_eq!(response.total_returned, 1);
assert_eq!(response.rows[0].cells["user_id"], json!("u1"));
assert_eq!(response.rows[0].cells["points"], json!(12));
}
#[test]
fn build_admin_database_table_row_normalizes_optional_sats_values() {
let row = build_admin_database_table_row(
&json!([[0, "u1"], [1, []]]),
&["user_id".to_string(), "deleted_at".to_string()],
);
assert_eq!(row.cells["user_id"], json!("u1"));
assert_eq!(row.cells["deleted_at"], json!(null));
}
#[test]
fn apply_admin_database_table_filters_supports_search_and_json_filters() {
let mut rows = vec![
build_admin_database_table_row(
&json!(["u1", "alice", 12]),
&[
"user_id".to_string(),
"name".to_string(),
"points".to_string(),
],
),
build_admin_database_table_row(
&json!(["u2", "bob", 8]),
&[
"user_id".to_string(),
"name".to_string(),
"points".to_string(),
],
),
];
apply_admin_database_table_filters(
&mut rows,
&AdminDatabaseTableRowsQuery {
search: Some("ali".to_string()),
filters: Some(r#"{"points":12}"#.to_string()),
limit: None,
},
)
.expect("filters should apply");
assert_eq!(rows.len(), 1);
assert_eq!(rows[0].cells["user_id"], json!("u1"));
}
#[test]
fn apply_admin_database_table_filters_rejects_non_object_filter() {
let mut rows = vec![build_admin_database_table_row(
&json!(["u1"]),
&["user_id".to_string()],
)];
let error = apply_admin_database_table_filters(
&mut rows,
&AdminDatabaseTableRowsQuery {
search: None,
filters: Some("[]".to_string()),
limit: None,
},
)
.expect_err("non object filter should fail");
assert_eq!(error.into_response().status(), StatusCode::BAD_REQUEST);
}
#[test]
fn build_admin_tracking_events_sql_quotes_filters_and_clamps_limit() {
let sql = build_admin_tracking_events_sql(&AdminTrackingEventListQuery {

18
server-rs/crates/api-server/src/app.rs
View File

@@ -14,8 +14,8 @@ use tracing::{Level, Span, error, info, info_span, warn};
use crate::{
admin::{
admin_debug_http, admin_list_tracking_events, admin_login, admin_me, admin_overview,
require_admin_auth,
admin_debug_http, admin_list_database_table_rows, admin_list_database_tables,
admin_list_tracking_events, admin_login, admin_me, admin_overview, require_admin_auth,
},
ai_tasks::{
append_ai_text_chunk, attach_ai_result_reference, cancel_ai_task, complete_ai_stage,
@@ -179,6 +179,20 @@ pub fn build_router(state: AppState) -> Router {
require_admin_auth,
)),
)
.route(
"/admin/api/database/tables",
get(admin_list_database_tables).route_layer(middleware::from_fn_with_state(
state.clone(),
require_admin_auth,
)),
)
.route(
"/admin/api/database/tables/{table_name}/rows",
get(admin_list_database_table_rows).route_layer(middleware::from_fn_with_state(
state.clone(),
require_admin_auth,
)),
)
.route(
"/admin/api/profile/redeem-codes",
get(admin_list_profile_redeem_codes)

36
server-rs/crates/shared-contracts/src/admin.rs
View File

@@ -77,6 +77,42 @@ pub struct AdminDatabaseTableStatPayload {
pub error_message: Option<String>,
}
// 后台表清单独立用于“表查询”页，避免页面必须先拉完整总览。
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct AdminDatabaseTableListResponse {
pub tables: Vec<String>,
pub fetch_errors: Vec<String>,
}
// 后台通用表查询参数，用户输入不进入 SQL，只在 API Server 内存中过滤。
#[derive(Clone, Debug, Serialize, Deserialize, Default, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct AdminDatabaseTableRowsQuery {
pub limit: Option<u32>,
pub search: Option<String>,
pub filters: Option<String>,
}
// 后台通用表查询响应，cells 使用列名映射，raw 保留原始行便于详情排障。
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
#[serde(rename_all = "camelCase")]
pub struct AdminDatabaseTableRowsResponse {
pub table_name: String,
pub columns: Vec<String>,
pub rows: Vec<AdminDatabaseTableRowPayload>,
pub total_returned: usize,
pub limit: u32,
}
// 单行查询结果，值统一用 JSON 承载以兼容不同表字段类型。
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
#[serde(rename_all = "camelCase")]
pub struct AdminDatabaseTableRowPayload {
pub cells: Value,
pub raw: Value,
}
// 调试请求只允许同源路径、受控请求头和有限请求体。
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
#[serde(rename_all = "camelCase")]