Merge pull request 'add row data inspect in admin' (#10) from hermes/hermes-f32d3246 into master

Reviewed-on: http://82.157.175.59:3000/GenarrativeAI/Genarrative/pulls/10

server-rs/crates/api-server/src/admin.rs

@@ -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 {

server-rs/crates/api-server/src/app.rs

@@ -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)