1use std::sync::Arc;
2
3use alopex_core::columnar::kvs_bridge::ColumnarKvsBridge;
4use alopex_core::columnar::segment_v2::SegmentMetaV2;
5use alopex_core::storage::format::bincode_config;
6use alopex_core::types::TxnMode;
7use alopex_core::{KVStore, KVTransaction};
8use alopex_sql::ast::ddl::VectorMetric;
9use alopex_sql::catalog::persistent::{PersistedTableMeta, TABLES_PREFIX};
10use alopex_sql::catalog::{
11 Catalog, CatalogError, CatalogOverlay, PersistentCatalog, TableMetadata,
12};
13use alopex_sql::planner::types::ResolvedType;
14use axum::extract::{Extension, Query};
15use axum::response::Response;
16use bincode::Options;
17use serde::{Deserialize, Serialize};
18
19use crate::error::{Result, ServerError};
20use crate::http::admin_api::{cluster_control_availability, ClusterControlAvailability};
21use crate::http::{error_response, json_response, RequestContext};
22use crate::server::ServerState;
23
24const DEFAULT_RESOURCE_LIMIT: usize = 50;
25const DEFAULT_COLUMNAR_COLUMN_LIMIT: usize = 20;
26const SYSTEM_PREFIXES: [&str; 6] = [
27 "__catalog__/",
28 "hnsw:",
29 "__alopex_",
30 "__alopex:",
31 "vector:",
32 "columnar:",
33];
34
35#[derive(Debug, Deserialize)]
36pub struct AdminResourcesQuery {
37 pub limit: Option<usize>,
38 pub include_columnar_columns: Option<bool>,
39 pub columnar_column_limit: Option<usize>,
40 pub kv_prefix: Option<String>,
41}
42
43#[derive(Debug, Serialize)]
44pub struct AdminResourcesResponse {
45 pub sql_tables: Vec<SqlTableResource>,
46 pub columnar_segments: Vec<ColumnarSegmentResource>,
47 pub kv_keys: Vec<String>,
48 pub truncated: TruncatedSections,
49 pub cluster_control: ClusterControlAvailability,
50}
51
52#[derive(Debug, Serialize)]
53pub struct TruncatedSections {
54 pub sql_tables: bool,
55 pub columnar_segments: bool,
56 pub kv_keys: bool,
57}
58
59#[derive(Debug, Serialize)]
60pub struct SqlTableResource {
61 pub name: String,
62 pub columns: Vec<SqlColumnResource>,
63}
64
65#[derive(Debug, Serialize)]
66pub struct SqlColumnResource {
67 pub name: String,
68 pub data_type: String,
69}
70
71#[derive(Debug, Serialize)]
72pub struct ColumnarSegmentResource {
73 pub id: String,
74 pub columns: Option<Vec<String>>,
75}
76
77pub async fn list(
78 Extension(state): Extension<Arc<ServerState>>,
79 Extension(ctx): Extension<RequestContext>,
80 Query(query): Query<AdminResourcesQuery>,
81) -> Response {
82 match list_impl(state.clone(), query) {
83 Ok(resp) => json_response(resp, state.config.max_response_size, &ctx),
84 Err(err) => error_response(err, &ctx),
85 }
86}
87
88fn list_impl(
89 state: Arc<ServerState>,
90 query: AdminResourcesQuery,
91) -> Result<AdminResourcesResponse> {
92 let limit = query.limit.unwrap_or(DEFAULT_RESOURCE_LIMIT);
93 let columnar_column_limit = query
94 .columnar_column_limit
95 .unwrap_or(DEFAULT_COLUMNAR_COLUMN_LIMIT);
96 let include_columnar_columns = query.include_columnar_columns.unwrap_or(false);
97
98 let (sql_tables, sql_truncated) =
99 list_sql_resources(state.store.clone(), state.catalog.clone(), limit)?;
100 let (columnar_segments, columnar_truncated) = list_columnar_resources(
101 state.store.clone(),
102 limit,
103 include_columnar_columns,
104 columnar_column_limit,
105 )?;
106 let (kv_keys, kv_truncated) = list_kv_resources(state.store.clone(), limit, query.kv_prefix)?;
107 let cluster = state.cluster_status_snapshot()?;
108 let cluster_control = cluster_control_availability(&cluster)?;
109
110 Ok(AdminResourcesResponse {
111 sql_tables,
112 columnar_segments,
113 kv_keys,
114 truncated: TruncatedSections {
115 sql_tables: sql_truncated,
116 columnar_segments: columnar_truncated,
117 kv_keys: kv_truncated,
118 },
119 cluster_control,
120 })
121}
122
123fn list_sql_resources(
124 store: Arc<alopex_core::kv::any::AnyKV>,
125 catalog: Arc<std::sync::RwLock<dyn Catalog + Send + Sync>>,
126 limit: usize,
127) -> Result<(Vec<SqlTableResource>, bool)> {
128 let mut tables = list_sql_resources_from_store(store.clone())?;
129 if tables.is_empty() {
130 tables = match PersistentCatalog::load(store.clone()) {
131 Ok(catalog) => {
132 let overlay = CatalogOverlay::new();
133 let mut tables = catalog.list_tables_in_txn("default", "default", &overlay);
134 if tables.is_empty() {
135 tables = catalog.list_tables_in_txn("main", "default", &overlay);
136 }
137 tables
138 }
139 Err(CatalogError::Kv(alopex_core::Error::NotFound)) => Vec::new(),
140 Err(err) => return Err(ServerError::Catalog(err)),
141 };
142 }
143 if tables.is_empty() {
144 let guard = catalog
145 .read()
146 .map_err(|_| ServerError::Internal("catalog lock poisoned".into()))?;
147 tables = guard.list_tables();
148 }
149
150 tables.sort_by(|a, b| a.name.cmp(&b.name));
151 let truncated = tables.len() > limit;
152 if tables.len() > limit {
153 tables.truncate(limit);
154 }
155
156 let resources = tables
157 .into_iter()
158 .map(|table| SqlTableResource {
159 name: table.name,
160 columns: table
161 .columns
162 .into_iter()
163 .map(|column| SqlColumnResource {
164 name: column.name,
165 data_type: resolved_type_to_string(&column.data_type),
166 })
167 .collect(),
168 })
169 .collect();
170
171 Ok((resources, truncated))
172}
173
174fn list_sql_resources_from_store(
175 store: Arc<alopex_core::kv::any::AnyKV>,
176) -> Result<Vec<TableMetadata>> {
177 let mut txn = store.begin(TxnMode::ReadOnly)?;
178 let mut tables = Vec::new();
179 for (_key, value) in txn.scan_prefix(TABLES_PREFIX)? {
180 let persisted: PersistedTableMeta = bincode_config()
181 .deserialize(&value)
182 .map_err(|err| ServerError::BadRequest(format!("catalog entry invalid: {err}")))?;
183 tables.push(TableMetadata::from(persisted));
184 }
185 txn.commit_self()?;
186 Ok(tables)
187}
188
189fn list_columnar_resources(
190 store: Arc<alopex_core::kv::any::AnyKV>,
191 limit: usize,
192 include_columns: bool,
193 columnar_column_limit: usize,
194) -> Result<(Vec<ColumnarSegmentResource>, bool)> {
195 let bridge = ColumnarKvsBridge::new(store);
196 let mut segments = bridge.list_segments().map_err(map_columnar_error)?;
197 segments.sort_by(|a, b| a.0.cmp(&b.0).then_with(|| a.1.cmp(&b.1)));
198 let truncated = segments.len() > limit;
199 if segments.len() > limit {
200 segments.truncate(limit);
201 }
202
203 let mut resources = Vec::with_capacity(segments.len());
204 for (table_id, segment_id) in segments {
205 let id = format!("{}:{}", table_id, segment_id);
206 let columns = if include_columns {
207 Some(list_columnar_columns(
208 &bridge,
209 table_id,
210 segment_id,
211 columnar_column_limit,
212 )?)
213 } else {
214 None
215 };
216 resources.push(ColumnarSegmentResource { id, columns });
217 }
218
219 Ok((resources, truncated))
220}
221
222fn list_columnar_columns(
223 bridge: &ColumnarKvsBridge,
224 table_id: u32,
225 segment_id: u64,
226 limit: usize,
227) -> Result<Vec<String>> {
228 let stats = bridge
229 .read_statistics(table_id, segment_id)
230 .map_err(map_columnar_error)?;
231 let meta: SegmentMetaV2 = bincode_config()
232 .deserialize(&stats)
233 .map_err(|err| ServerError::BadRequest(format!("columnar segment invalid: {err}")))?;
234 let mut names: Vec<String> = meta
235 .schema
236 .columns
237 .into_iter()
238 .map(|column| column.name)
239 .collect();
240 if names.len() > limit {
241 names.truncate(limit);
242 }
243 Ok(names)
244}
245
246fn list_kv_resources(
247 store: Arc<alopex_core::kv::any::AnyKV>,
248 limit: usize,
249 kv_prefix: Option<String>,
250) -> Result<(Vec<String>, bool)> {
251 let prefix = kv_prefix.unwrap_or_default();
252 let mut txn = store.begin(TxnMode::ReadOnly)?;
253 let mut keys = Vec::new();
254 let mut truncated = false;
255 for (key, _) in txn.scan_prefix(prefix.as_bytes())? {
256 if is_system_key(&key) {
257 continue;
258 }
259 let Ok(key_str) = std::str::from_utf8(&key) else {
260 continue;
261 };
262 if key_str.is_empty() || key_str.chars().any(|ch| ch.is_control()) {
263 continue;
264 }
265 if keys.len() < limit {
266 keys.push(key_str.to_string());
267 } else {
268 truncated = true;
269 break;
270 }
271 }
272 txn.commit_self()?;
273 Ok((keys, truncated))
274}
275
276fn is_system_key(key: &[u8]) -> bool {
277 SYSTEM_PREFIXES
278 .iter()
279 .any(|prefix| key.starts_with(prefix.as_bytes()))
280}
281
282fn resolved_type_to_string(resolved_type: &ResolvedType) -> String {
283 match resolved_type {
284 ResolvedType::Integer => "INTEGER".to_string(),
285 ResolvedType::BigInt => "BIGINT".to_string(),
286 ResolvedType::Float => "FLOAT".to_string(),
287 ResolvedType::Double => "DOUBLE".to_string(),
288 ResolvedType::Text => "TEXT".to_string(),
289 ResolvedType::Blob => "BLOB".to_string(),
290 ResolvedType::Boolean => "BOOLEAN".to_string(),
291 ResolvedType::Timestamp => "TIMESTAMP".to_string(),
292 ResolvedType::Vector { dimension, metric } => {
293 let metric = match metric {
294 VectorMetric::Cosine => "COSINE",
295 VectorMetric::L2 => "L2",
296 VectorMetric::Inner => "INNER",
297 };
298 format!("VECTOR({dimension}, {metric})")
299 }
300 ResolvedType::Null => "NULL".to_string(),
301 }
302}
303
304fn map_columnar_error(err: alopex_core::columnar::ColumnarError) -> ServerError {
305 match err {
306 alopex_core::columnar::ColumnarError::NotFound => {
307 ServerError::NotFound("columnar segment not found".into())
308 }
309 alopex_core::columnar::ColumnarError::InvalidFormat(message) => {
310 ServerError::BadRequest(format!("columnar segment invalid: {message}"))
311 }
312 alopex_core::columnar::ColumnarError::MemoryLimitExceeded { limit, requested } => {
313 ServerError::PayloadTooLarge(format!(
314 "memory limit {limit} exceeded by {requested} bytes"
315 ))
316 }
317 other => ServerError::Core(other.into()),
318 }
319}