use crate::daemon;
use crate::error::{ErrorCode, McpError};
use crate::schema::ColumnSchema;
use hyperdb_api::{
escape_sql_path, Catalog, Connection, CreateMode, HyperProcess, Parameters, SqlType,
};
use serde_json::{json, Value};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
static EPHEMERAL_SEQ: AtomicU64 = AtomicU64::new(0);
const PERSISTENT_ALIAS: &str = "persistent";
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct PersistentAttachOutcome {
pub file_was_created: bool,
}
fn attach_default_persistent(
connection: &Connection,
persistent_path: &Path,
primary_db_name: &str,
) -> Result<PersistentAttachOutcome, McpError> {
let path_str = persistent_path.to_string_lossy();
let file_was_created = !persistent_path.exists();
if file_was_created {
let create_sql = format!(
"CREATE DATABASE IF NOT EXISTS {}",
escape_sql_path(&path_str)
);
connection.execute_command(&create_sql).map_err(|e| {
McpError::new(
ErrorCode::InternalError,
format!("Failed to create persistent database: {e}"),
)
})?;
}
let attach_sql = format!(
"ATTACH DATABASE {path} AS \"{alias}\"",
path = escape_sql_path(&path_str),
alias = PERSISTENT_ALIAS,
);
connection.execute_command(&attach_sql).map_err(|e| {
McpError::new(
ErrorCode::InternalError,
format!("Failed to attach persistent database: {e}"),
)
})?;
let pin_sql = format!(
"SET schema_search_path = '{}'",
primary_db_name.replace('\'', "''")
);
connection.execute_command(&pin_sql).map_err(|e| {
McpError::new(
ErrorCode::InternalError,
format!("Failed to pin schema_search_path: {e}"),
)
})?;
Ok(PersistentAttachOutcome { file_was_created })
}
fn path_stem(path: &Path) -> String {
path.file_stem()
.and_then(|s| s.to_str())
.unwrap_or("scratch")
.to_string()
}
#[derive(Debug)]
pub struct ScopedSearchPath<'a> {
engine: &'a Engine,
restore_to: String,
}
impl Drop for ScopedSearchPath<'_> {
fn drop(&mut self) {
let sql = format!(
"SET schema_search_path = '{}'",
self.restore_to.replace('\'', "''")
);
if let Err(e) = self.engine.execute_command(&sql) {
tracing::warn!(
error = %e.message,
"failed to restore schema_search_path — next tool call may route incorrectly"
);
}
}
}
#[derive(Debug)]
pub struct Engine {
hyper: Option<HyperProcess>,
daemon_endpoint: Option<String>,
daemon_health_port: Option<u16>,
connection: Connection,
ephemeral_path: PathBuf,
persistent_path: Option<PathBuf>,
persistent_was_created: bool,
catalog_present_cache: std::sync::Mutex<std::collections::HashMap<String, bool>>,
log_dir: PathBuf,
}
impl Engine {
pub fn new(persistent_db_path: Option<String>) -> Result<Self, McpError> {
Self::new_with_mode(persistent_db_path, false)
}
pub fn new_no_daemon(persistent_db_path: Option<String>) -> Result<Self, McpError> {
Self::new_with_mode(persistent_db_path, true)
}
#[expect(
clippy::needless_pass_by_value,
reason = "Option<String> is consumed by the path-expansion logic below"
)]
fn new_with_mode(
persistent_db_path: Option<String>,
no_daemon: bool,
) -> Result<Self, McpError> {
let persistent_path = match persistent_db_path.as_deref() {
Some(p) => {
let path = PathBuf::from(shellexpand_tilde(p));
if let Some(parent) = path.parent() {
std::fs::create_dir_all(parent).map_err(|e| {
McpError::new(
ErrorCode::PermissionDenied,
format!("Cannot create persistent-db directory: {e}"),
)
})?;
}
Some(path)
}
None => None,
};
let seq = EPHEMERAL_SEQ.fetch_add(1, Ordering::Relaxed);
let ephemeral_dir =
std::env::temp_dir().join(format!("hyperdb-mcp-{}-{seq}", std::process::id()));
std::fs::create_dir_all(&ephemeral_dir).map_err(|e| {
McpError::new(
ErrorCode::InternalError,
format!("Cannot create ephemeral directory: {e}"),
)
})?;
let ephemeral_path = ephemeral_dir.join("scratch.hyper");
let log_dir = resolve_log_dir(persistent_db_path.as_deref());
std::fs::create_dir_all(&log_dir).map_err(|e| {
McpError::new(
ErrorCode::PermissionDenied,
format!("Cannot create log directory {}: {e}", log_dir.display()),
)
})?;
if !no_daemon {
if let Some(engine) =
Self::try_daemon_mode(&ephemeral_path, persistent_path.clone(), &log_dir)?
{
return Ok(engine);
}
}
let mut params = Parameters::new();
params.set("log_file_max_count", "2");
params.set("log_file_size_limit", "100M");
params.set("log_dir", log_dir.to_string_lossy().as_ref());
let hyper = HyperProcess::new(None, Some(¶ms)).map_err(|e| {
let msg = e.to_string();
if msg.contains("hyperd") || msg.contains("HYPERD_PATH") || msg.contains("No such file")
{
McpError::new(ErrorCode::HyperdNotFound, msg)
} else {
McpError::new(ErrorCode::InternalError, msg)
}
})?;
let connection = Connection::new(&hyper, &ephemeral_path, CreateMode::CreateAndReplace)
.map_err(|e| {
McpError::new(ErrorCode::InternalError, format!("Failed to connect: {e}"))
})?;
bootstrap_public_schema(&connection)?;
let primary_db_name = path_stem(&ephemeral_path);
let persistent_was_created = Self::attach_persistent_if_present(
&connection,
persistent_path.as_deref(),
&primary_db_name,
)?;
Ok(Self {
hyper: Some(hyper),
daemon_endpoint: None,
daemon_health_port: None,
connection,
ephemeral_path,
persistent_path,
persistent_was_created,
catalog_present_cache: std::sync::Mutex::new(std::collections::HashMap::new()),
log_dir,
})
}
fn attach_persistent_if_present(
connection: &Connection,
persistent_path: Option<&Path>,
primary_db_name: &str,
) -> Result<bool, McpError> {
let Some(path) = persistent_path else {
return Ok(false);
};
let outcome = attach_default_persistent(connection, path, primary_db_name)?;
Ok(outcome.file_was_created)
}
fn try_daemon_mode(
ephemeral_path: &Path,
persistent_path: Option<PathBuf>,
log_dir: &Path,
) -> Result<Option<Self>, McpError> {
let info = match daemon::spawn::ensure_daemon(daemon::discovery::resolve_port_scan()) {
Ok(info) => info,
Err(e) => {
tracing::debug!(error = %e, "daemon unavailable, falling back to local mode");
return Ok(None);
}
};
let endpoint = &info.hyperd_endpoint;
let connection = Connection::connect(
endpoint,
&ephemeral_path.to_string_lossy(),
CreateMode::CreateAndReplace,
)
.map_err(|e| {
daemon::health::report_hyperd_error_to_daemon();
McpError::new(
ErrorCode::InternalError,
format!("Failed to connect to daemon hyperd at {endpoint}: {e}"),
)
})?;
bootstrap_public_schema(&connection)?;
let _ = daemon::health::send_command(info.health_port, "HEARTBEAT");
let primary_db_name = path_stem(ephemeral_path);
let persistent_was_created = Self::attach_persistent_if_present(
&connection,
persistent_path.as_deref(),
&primary_db_name,
)?;
Ok(Some(Self {
hyper: None,
daemon_endpoint: Some(info.hyperd_endpoint),
daemon_health_port: Some(info.health_port),
connection,
ephemeral_path: ephemeral_path.to_path_buf(),
persistent_path,
persistent_was_created,
catalog_present_cache: std::sync::Mutex::new(std::collections::HashMap::new()),
log_dir: log_dir.to_path_buf(),
}))
}
pub fn is_running(&self) -> bool {
if let Some(ref hyper) = self.hyper {
hyper.is_running()
} else if let Some(ref endpoint) = self.daemon_endpoint {
probe_endpoint_alive(endpoint)
} else {
daemon::discovery::discover().is_some()
}
}
pub fn hyperd_endpoint(&self) -> Result<String, McpError> {
if let Some(ref endpoint) = self.daemon_endpoint {
return Ok(endpoint.clone());
}
self.hyper
.as_ref()
.ok_or_else(|| McpError::new(ErrorCode::InternalError, "no hyperd endpoint available"))?
.require_endpoint()
.map(std::string::ToString::to_string)
.map_err(|e| McpError::new(ErrorCode::InternalError, e.to_string()))
}
pub fn daemon_health_port(&self) -> Option<u16> {
self.daemon_health_port
}
pub fn ephemeral_path(&self) -> &Path {
&self.ephemeral_path
}
pub fn persistent_path(&self) -> Option<&Path> {
self.persistent_path.as_deref()
}
pub const PERSISTENT_ALIAS: &'static str = "persistent";
pub fn primary_db_name(&self) -> String {
self.ephemeral_path
.file_stem()
.and_then(|s| s.to_str())
.unwrap_or("scratch")
.to_string()
}
pub fn resolve_target_db(&self, requested: Option<&str>) -> Result<String, McpError> {
match requested.map(str::trim) {
None | Some("") => Ok(self.primary_db_name()),
Some(other) if other.eq_ignore_ascii_case(Self::PERSISTENT_ALIAS) => {
if self.persistent_path.is_none() {
return Err(McpError::new(
ErrorCode::InvalidArgument,
"no persistent database in this session — \
hyperdb-mcp was started with --ephemeral-only"
.to_string(),
));
}
Ok(Self::PERSISTENT_ALIAS.to_string())
}
Some(other) => Ok(other.to_ascii_lowercase()),
}
}
pub fn scoped_search_path(&self, alias: &str) -> Result<ScopedSearchPath<'_>, McpError> {
let primary = self.primary_db_name();
let set_sql = format!("SET schema_search_path = '{}'", alias.replace('\'', "''"));
self.execute_command(&set_sql)?;
Ok(ScopedSearchPath {
engine: self,
restore_to: primary,
})
}
pub fn log_dir(&self) -> &Path {
&self.log_dir
}
pub fn hyperd_log_path(&self) -> Option<PathBuf> {
let entries = std::fs::read_dir(&self.log_dir).ok()?;
let mut candidates: Vec<(std::time::SystemTime, PathBuf)> = entries
.filter_map(std::result::Result::ok)
.filter_map(|e| {
let path = e.path();
let name = path.file_name()?.to_str()?;
if name.starts_with("hyperd")
&& std::path::Path::new(name)
.extension()
.is_some_and(|ext| ext.eq_ignore_ascii_case("log"))
{
let mtime = e.metadata().ok().and_then(|m| m.modified().ok())?;
Some((mtime, path))
} else {
None
}
})
.collect();
candidates.sort_by_key(|b| std::cmp::Reverse(b.0));
candidates.into_iter().next().map(|(_, p)| p)
}
pub fn has_persistent(&self) -> bool {
self.persistent_path.is_some()
}
pub fn persistent_was_just_created(&self) -> bool {
self.persistent_was_created
}
pub fn catalog_present_in<F>(&self, alias: &str, prober: F) -> Result<bool, McpError>
where
F: Fn(&Engine) -> Result<bool, McpError>,
{
let key = alias.to_ascii_lowercase();
if let Ok(guard) = self.catalog_present_cache.lock() {
if let Some(&present) = guard.get(&key) {
return Ok(present);
}
}
let present = prober(self)?;
if let Ok(mut guard) = self.catalog_present_cache.lock() {
guard.insert(key, present);
}
Ok(present)
}
pub fn mark_catalog_present_for(&self, alias: &str) {
let key = alias.to_ascii_lowercase();
if let Ok(mut guard) = self.catalog_present_cache.lock() {
guard.insert(key, true);
}
}
pub fn clear_catalog_cache_for(&self, alias: &str) {
let key = alias.to_ascii_lowercase();
if let Ok(mut guard) = self.catalog_present_cache.lock() {
guard.remove(&key);
}
}
pub fn connection(&self) -> &Connection {
&self.connection
}
pub fn execute_command(&self, sql: &str) -> Result<u64, McpError> {
self.connection.execute_command(sql).map_err(McpError::from)
}
#[allow(
deprecated,
reason = "Engine borrows &self; the RAII guard requires &mut. Migration tracked in issue #72."
)]
pub fn execute_in_transaction<F, T>(&self, f: F) -> Result<T, McpError>
where
F: FnOnce(&Engine) -> Result<T, McpError>,
{
self.connection
.begin_transaction()
.map_err(McpError::from)?;
tracing::debug!("tx: BEGIN issued");
let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| f(self)));
match result {
Ok(Ok(val)) => {
tracing::debug!("tx: closure returned Ok, issuing COMMIT");
self.connection.commit().map_err(McpError::from)?;
Ok(val)
}
Ok(Err(e)) => {
tracing::debug!(err = %e, "tx: closure returned Err, issuing ROLLBACK");
if let Err(rb_err) = self.connection.rollback() {
tracing::warn!(
"rollback after error failed (original error preserved): {}",
rb_err
);
} else {
tracing::debug!("tx: ROLLBACK succeeded");
}
Err(e)
}
Err(panic_payload) => {
tracing::error!("tx: closure panicked, issuing ROLLBACK before resuming unwind");
let _ = self.connection.rollback();
std::panic::resume_unwind(panic_payload)
}
}
}
pub fn execute_query_to_json(&self, sql: &str) -> Result<Vec<Value>, McpError> {
let mut result = self.connection.execute_query(sql).map_err(McpError::from)?;
let mut rows_json = Vec::new();
let mut schema_opt = None;
while let Some(chunk) = result.next_chunk().map_err(McpError::from)? {
if schema_opt.is_none() {
schema_opt = result.schema();
}
if let Some(ref schema) = schema_opt {
let columns = schema.columns();
for row in &chunk {
let mut obj = serde_json::Map::new();
for col in columns {
let val = row_value_to_json(row, col.index(), &col.sql_type());
obj.insert(col.name().to_string(), val);
}
rows_json.push(Value::Object(obj));
}
}
}
Ok(rows_json)
}
pub fn create_table(
&self,
table_name: &str,
columns: &[ColumnSchema],
replace: bool,
) -> Result<(), McpError> {
self.create_table_in(table_name, columns, replace, None)
}
pub fn create_table_in(
&self,
table_name: &str,
columns: &[ColumnSchema],
replace: bool,
target_db: Option<&str>,
) -> Result<(), McpError> {
if columns.is_empty() {
return Err(McpError::new(
ErrorCode::EmptyData,
"No columns to create table from",
));
}
for col in columns {
if crate::schema::map_hyper_type(&col.hyper_type).is_none() {
return Err(McpError::new(
ErrorCode::SchemaMismatch,
format!(
"Unknown type '{}' for column '{}'",
col.hyper_type, col.name
),
));
}
}
let quoted_table = match target_db {
Some(db) => {
let esc_db = db.replace('"', "\"\"");
let esc_tbl = table_name.replace('"', "\"\"");
format!("\"{esc_db}\".\"public\".\"{esc_tbl}\"")
}
None => format!("\"{}\"", table_name.replace('"', "\"\"")),
};
if replace {
self.connection
.execute_command(&format!("DROP TABLE IF EXISTS {quoted_table}"))
.map_err(McpError::from)?;
}
let col_defs: Vec<String> = columns
.iter()
.map(|c| {
let nullable = if c.nullable { "" } else { " NOT NULL" };
format!(
"\"{}\" {}{}",
c.name.replace('"', "\"\""),
c.hyper_type,
nullable
)
})
.collect();
let create_sql = format!(
"CREATE TABLE IF NOT EXISTS {} ({})",
quoted_table,
col_defs.join(", ")
);
self.connection
.execute_command(&create_sql)
.map_err(McpError::from)?;
Ok(())
}
pub fn column_metadata(&self, table: &str) -> Result<Vec<ColumnSchema>, McpError> {
let catalog = Catalog::new(&self.connection);
let def = catalog
.get_table_definition(table)
.map_err(McpError::from)?;
Ok(def
.columns()
.iter()
.map(|c| ColumnSchema {
name: c.name.clone(),
hyper_type: c.type_name().to_string(),
nullable: c.nullable,
})
.collect())
}
pub fn column_metadata_in(
&self,
target_db: Option<&str>,
table: &str,
) -> Result<Vec<ColumnSchema>, McpError> {
let Some(db) = target_db else {
return self.column_metadata(table);
};
let rows = describe_columns_via_pg_catalog(self, db, table)?;
if rows.is_empty() {
return Err(McpError::new(
ErrorCode::TableNotFound,
format!("Table '{table}' does not exist in database '{db}'"),
));
}
Ok(rows
.into_iter()
.map(|r| ColumnSchema {
name: r
.get("name")
.and_then(|v| v.as_str())
.unwrap_or_default()
.to_string(),
hyper_type: r
.get("type")
.and_then(|v| v.as_str())
.unwrap_or_default()
.to_string(),
nullable: r
.get("nullable")
.and_then(serde_json::Value::as_bool)
.unwrap_or(true),
})
.collect())
}
pub fn table_exists(&self, table: &str) -> Result<bool, McpError> {
let catalog = Catalog::new(&self.connection);
let names = catalog.get_table_names("public").map_err(McpError::from)?;
Ok(names.iter().any(|n| n.as_str() == table))
}
pub fn table_exists_in(&self, target_db: Option<&str>, table: &str) -> Result<bool, McpError> {
let Some(db) = target_db else {
return self.table_exists(table);
};
let esc_db = db.replace('"', "\"\"");
let esc_tbl = table.replace('\'', "''");
let sql = format!(
"SELECT 1 AS one FROM \"{esc_db}\".pg_catalog.pg_tables \
WHERE schemaname = 'public' AND tablename = '{esc_tbl}'"
);
let rows = self.execute_query_to_json(&sql)?;
Ok(!rows.is_empty())
}
pub fn alter_table_add_columns(
&self,
table: &str,
cols: &[ColumnSchema],
) -> Result<(), McpError> {
self.alter_table_add_columns_in(None, table, cols)
}
pub fn alter_table_add_columns_in(
&self,
target_db: Option<&str>,
table: &str,
cols: &[ColumnSchema],
) -> Result<(), McpError> {
if cols.is_empty() {
return Ok(());
}
for col in cols {
if crate::schema::map_hyper_type(&col.hyper_type).is_none() {
return Err(McpError::new(
ErrorCode::SchemaMismatch,
format!(
"Unknown type '{}' for column '{}'",
col.hyper_type, col.name
),
));
}
}
let quoted_table = match target_db {
Some(db) => {
let esc_db = db.replace('"', "\"\"");
let esc_tbl = table.replace('"', "\"\"");
format!("\"{esc_db}\".\"public\".\"{esc_tbl}\"")
}
None => format!("\"{}\"", table.replace('"', "\"\"")),
};
let add_clauses = cols
.iter()
.map(|c| {
format!(
"ADD COLUMN \"{}\" {}",
c.name.replace('"', "\"\""),
c.hyper_type
)
})
.collect::<Vec<_>>()
.join(", ");
let sql = format!("ALTER TABLE {quoted_table} {add_clauses}");
self.connection
.execute_command(&sql)
.map_err(McpError::from)?;
Ok(())
}
pub fn describe_tables(&self) -> Result<Vec<Value>, McpError> {
let catalog = Catalog::new(&self.connection);
let table_names = catalog.get_table_names("public").map_err(McpError::from)?;
let mut tables = Vec::new();
for name in &table_names {
if is_internal_table(name.as_str()) {
continue;
}
tables.push(describe_table_with_catalog(&catalog, name.as_str())?);
}
Ok(tables)
}
pub fn describe_table(&self, table_name: &str) -> Result<Value, McpError> {
if is_internal_table(table_name) {
return Err(McpError::new(
ErrorCode::TableNotFound,
format!("Table '{table_name}' does not exist"),
));
}
let catalog = Catalog::new(&self.connection);
let exists = catalog
.get_table_names("public")
.map_err(McpError::from)?
.iter()
.any(|n| n.as_str() == table_name);
if !exists {
return Err(McpError::new(
ErrorCode::TableNotFound,
format!("Table '{table_name}' does not exist"),
));
}
describe_table_with_catalog(&catalog, table_name)
}
pub fn sample_table(&self, table_name: &str, n: u64) -> Result<Value, McpError> {
self.sample_table_in(None, table_name, n)
}
pub fn sample_table_in(
&self,
target_db: Option<&str>,
table_name: &str,
n: u64,
) -> Result<Value, McpError> {
let n = n.clamp(1, 100);
let qualified = match target_db {
Some(db) => {
let esc_db = db.replace('"', "\"\"");
let esc_tbl = table_name.replace('"', "\"\"");
format!("\"{esc_db}\".\"public\".\"{esc_tbl}\"")
}
None => format!("\"{}\"", table_name.replace('"', "\"\"")),
};
let select_sql = format!("SELECT * FROM {qualified} LIMIT {n}");
let rows = match self.execute_query_to_json(&select_sql) {
Ok(r) => r,
Err(e) => return Err(translate_table_missing(e, table_name)),
};
let count_sql = format!("SELECT COUNT(*) AS cnt FROM {qualified}");
let row_count = self
.execute_query_to_json(&count_sql)
.ok()
.and_then(|rs| {
rs.first()
.and_then(|r| r.get("cnt").and_then(serde_json::Value::as_i64))
})
.unwrap_or(0);
let columns: Vec<Value> = match target_db {
None => {
let catalog = Catalog::new(&self.connection);
catalog
.get_table_definition(table_name)
.map(|def| {
def.columns()
.iter()
.map(|col| {
json!({
"name": col.name,
"type": col.type_name(),
"nullable": col.nullable,
})
})
.collect()
})
.unwrap_or_default()
}
Some(db) => describe_columns_via_pg_catalog(self, db, table_name).unwrap_or_default(),
};
Ok(json!({
"table": table_name,
"row_count": row_count,
"sample_size": rows.len(),
"schema": columns,
"rows": rows,
}))
}
pub fn describe_tables_in(&self, target_db: Option<&str>) -> Result<Vec<Value>, McpError> {
match target_db {
None => self.describe_tables(),
Some(db) => {
let esc_db = db.replace('"', "\"\"");
let list_sql = format!(
"SELECT tablename FROM \"{esc_db}\".pg_catalog.pg_tables \
WHERE schemaname = 'public' ORDER BY tablename"
);
let names_rows = self.execute_query_to_json(&list_sql)?;
let mut out = Vec::new();
for row in &names_rows {
let Some(name) = row.get("tablename").and_then(|v| v.as_str()) else {
continue;
};
if is_internal_table(name) {
continue;
}
out.push(self.describe_table_in(Some(db), name)?);
}
Ok(out)
}
}
}
pub fn describe_table_in(
&self,
target_db: Option<&str>,
table_name: &str,
) -> Result<Value, McpError> {
if is_internal_table(table_name) {
return Err(McpError::new(
ErrorCode::TableNotFound,
format!("Table '{table_name}' does not exist"),
));
}
match target_db {
None => self.describe_table(table_name),
Some(db) => {
let esc_db = db.replace('"', "\"\"");
let esc_tbl = table_name.replace('\'', "''");
let exists_sql = format!(
"SELECT 1 FROM \"{esc_db}\".pg_catalog.pg_tables \
WHERE schemaname = 'public' AND tablename = '{esc_tbl}'"
);
let rows = self.execute_query_to_json(&exists_sql)?;
if rows.is_empty() {
return Err(McpError::new(
ErrorCode::TableNotFound,
format!("Table '{table_name}' does not exist in database '{db}'"),
));
}
let columns = describe_columns_via_pg_catalog(self, db, table_name)?;
let qualified = format!(
"\"{esc_db}\".\"public\".\"{}\"",
table_name.replace('"', "\"\"")
);
let count_sql = format!("SELECT COUNT(*) AS cnt FROM {qualified}");
let row_count = self
.execute_query_to_json(&count_sql)
.ok()
.and_then(|rs| {
rs.first()
.and_then(|r| r.get("cnt").and_then(serde_json::Value::as_i64))
})
.unwrap_or(0);
Ok(json!({
"name": table_name,
"row_count": row_count,
"columns": columns,
}))
}
}
}
pub fn status(&self) -> Result<Value, McpError> {
let catalog = Catalog::new(&self.connection);
let all_names = catalog.get_table_names("public").map_err(McpError::from)?;
let table_names: Vec<_> = all_names
.iter()
.filter(|n| !is_internal_table(n.as_str()))
.collect();
let table_count = table_names.len();
let total_rows: i64 = table_names
.iter()
.map(|name| catalog.get_row_count(name.as_str()).unwrap_or(0))
.sum();
let ephemeral_bytes = std::fs::metadata(&self.ephemeral_path).map_or(0, |m| m.len());
let persistent_bytes = self
.persistent_path
.as_ref()
.and_then(|p| std::fs::metadata(p).ok())
.map_or(0u64, |m| m.len());
let disk_bytes = ephemeral_bytes.saturating_add(persistent_bytes);
let hyperd_log = self.hyperd_log_path().map_or(Value::Null, |p| {
Value::String(p.to_string_lossy().into_owned())
});
let client_log_path = self.log_dir.join(CLIENT_LOG_FILE_NAME);
let client_log = if client_log_path.exists() {
Value::String(client_log_path.to_string_lossy().into_owned())
} else {
Value::Null
};
let persistent_path_value = self.persistent_path.as_ref().map_or(Value::Null, |p| {
Value::String(p.to_string_lossy().into_owned())
});
let in_daemon_mode = self.daemon_endpoint.is_some();
let endpoint_value = self.hyperd_endpoint().map_or(Value::Null, Value::String);
let health_port_value = self
.daemon_health_port
.map_or(Value::Null, |p| Value::Number(p.into()));
Ok(json!({
"hyperd_running": self.is_running(),
"ephemeral_path": self.ephemeral_path.to_string_lossy(),
"persistent_path": persistent_path_value,
"has_persistent": self.has_persistent(),
"table_count": table_count,
"total_rows": total_rows,
"disk_usage_bytes": disk_bytes,
"engine": {
"mode": if in_daemon_mode { "daemon" } else { "local" },
"hyperd_endpoint": endpoint_value,
"daemon_health_port": health_port_value,
},
"hyper_rust_api_version": crate::version::hyper_api_version_string(),
"logs": {
"log_dir": self.log_dir.to_string_lossy(),
"hyperd_log": hyperd_log,
"client_log": client_log,
},
}))
}
}
fn row_value_to_json(row: &hyperdb_api::Row, idx: usize, sql_type: &SqlType) -> Value {
use hyperdb_api::oids;
use hyperdb_api::{Date, Numeric, OffsetTimestamp, Timestamp};
if row.is_null(idx) {
return Value::Null;
}
let oid_val = sql_type.internal_oid();
if oid_val == oids::BOOL.0 {
return row.get::<bool>(idx).map_or(Value::Null, Value::Bool);
}
if oid_val == oids::SMALL_INT.0 {
return row
.get::<i16>(idx)
.map_or(Value::Null, |v| Value::Number(v.into()));
}
if oid_val == oids::INT.0 {
return row
.get::<i32>(idx)
.map_or(Value::Null, |v| Value::Number(v.into()));
}
if oid_val == oids::BIG_INT.0 {
return row
.get::<i64>(idx)
.map_or(Value::Null, |v| Value::Number(v.into()));
}
if oid_val == oids::DOUBLE.0 || oid_val == oids::FLOAT.0 {
return row
.get::<f64>(idx)
.and_then(|v| serde_json::Number::from_f64(v).map(Value::Number))
.unwrap_or(Value::Null);
}
if oid_val == oids::NUMERIC.0 {
return row.get::<Numeric>(idx).map_or(Value::Null, |n| {
let s = n.to_string();
s.parse::<f64>()
.ok()
.and_then(serde_json::Number::from_f64)
.map(Value::Number)
.unwrap_or(Value::String(s))
});
}
if oid_val == oids::DATE.0 {
return row
.get::<Date>(idx)
.map_or(Value::Null, |d| Value::String(d.to_string()));
}
if oid_val == oids::TIMESTAMP.0 {
return row
.get::<Timestamp>(idx)
.map_or(Value::Null, |t| Value::String(t.to_string()));
}
if oid_val == oids::TIMESTAMP_TZ.0 {
return row
.get::<OffsetTimestamp>(idx)
.map_or(Value::Null, |t| Value::String(t.to_string()));
}
if oid_val == oids::TEXT.0 || oid_val == oids::VARCHAR.0 {
return row.get::<String>(idx).map_or(Value::Null, Value::String);
}
row.get::<String>(idx).map_or(Value::Null, Value::String)
}
pub const CLIENT_LOG_FILE_NAME: &str = "hyperdb-mcp.log";
pub const HYPERDB_INTERNAL_PREFIX: &str = "_hyperdb_";
#[must_use]
pub fn is_internal_table(name: &str) -> bool {
name.starts_with(HYPERDB_INTERNAL_PREFIX)
}
#[must_use]
pub fn resolve_log_dir(persistent_db_path: Option<&str>) -> PathBuf {
match persistent_db_path {
Some(p) => {
let expanded = PathBuf::from(shellexpand_tilde(p));
expanded
.parent()
.map_or_else(|| PathBuf::from("."), std::path::Path::to_path_buf)
}
None => std::env::temp_dir().join(format!("hyperdb-mcp-{}", std::process::id())),
}
}
fn describe_columns_via_pg_catalog(
engine: &Engine,
db_alias: &str,
table_name: &str,
) -> Result<Vec<Value>, McpError> {
let esc_db = db_alias.replace('"', "\"\"");
let esc_tbl = table_name.replace('\'', "''");
let sql = format!(
"SELECT a.attname AS name, \
t.typname AS type_name, \
NOT a.attnotnull AS nullable, \
a.attnum AS ordinal \
FROM \"{esc_db}\".pg_catalog.pg_attribute a \
JOIN \"{esc_db}\".pg_catalog.pg_class c ON a.attrelid = c.oid \
JOIN \"{esc_db}\".pg_catalog.pg_namespace n ON c.relnamespace = n.oid \
JOIN \"{esc_db}\".pg_catalog.pg_type t ON a.atttypid = t.oid \
WHERE n.nspname = 'public' \
AND c.relname = '{esc_tbl}' \
AND a.attnum > 0 \
ORDER BY a.attnum"
);
let rows = engine.execute_query_to_json(&sql)?;
Ok(rows
.into_iter()
.map(|r| {
json!({
"name": r.get("name").cloned().unwrap_or(Value::Null),
"type": r.get("type_name").cloned().unwrap_or(Value::Null),
"nullable": r.get("nullable").cloned().unwrap_or(Value::Bool(true)),
})
})
.collect())
}
fn describe_table_with_catalog(catalog: &Catalog<'_>, name: &str) -> Result<Value, McpError> {
let def = catalog.get_table_definition(name).map_err(McpError::from)?;
let row_count = catalog.get_row_count(name).unwrap_or(0);
let columns: Vec<Value> = def
.columns()
.iter()
.map(|col| {
json!({
"name": col.name,
"type": col.type_name(),
"nullable": col.nullable,
})
})
.collect();
Ok(json!({
"name": name,
"columns": columns,
"row_count": row_count,
}))
}
fn translate_table_missing(err: McpError, table_name: &str) -> McpError {
let m = err.message.to_lowercase();
let looks_like_missing = m.contains("does not exist")
|| m.contains("relation")
|| m.contains("undefined table")
|| err.message.contains("42P01");
if looks_like_missing {
McpError::new(
ErrorCode::TableNotFound,
format!("Table '{table_name}' does not exist"),
)
} else {
err
}
}
#[must_use]
pub fn is_read_only_sql(sql: &str) -> bool {
matches!(classify_statement(sql), StatementKind::ReadOnly)
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum StatementKind {
ReadOnly,
Ddl,
Dml,
TransactionControl,
Other,
}
#[must_use]
pub fn classify_statement(sql: &str) -> StatementKind {
let stripped = strip_leading_sql_comments(sql);
let first_token: String = stripped
.chars()
.take_while(|c| c.is_alphabetic())
.flat_map(char::to_uppercase)
.collect();
match first_token.as_str() {
"SELECT" | "WITH" | "EXPLAIN" | "SHOW" | "VALUES" => StatementKind::ReadOnly,
"CREATE" | "DROP" | "ALTER" | "TRUNCATE" | "RENAME" => StatementKind::Ddl,
"INSERT" | "UPDATE" | "DELETE" | "COPY" | "MERGE" => StatementKind::Dml,
"BEGIN" | "START" | "COMMIT" | "END" | "ROLLBACK" | "ABORT" | "SAVEPOINT" | "RELEASE" => {
StatementKind::TransactionControl
}
_ => StatementKind::Other,
}
}
pub(crate) fn strip_leading_sql_comments(sql: &str) -> &str {
let mut s = sql;
loop {
s = s.trim_start();
if s.starts_with("--") {
match s.find(&['\n', '\r'][..]) {
Some(pos) => {
let mut next = pos + 1;
if s.as_bytes().get(pos) == Some(&b'\r')
&& s.as_bytes().get(pos + 1) == Some(&b'\n')
{
next = pos + 2;
}
s = &s[next..];
}
None => return "",
}
} else if s.starts_with("/*") {
let mut depth = 0u32;
let mut chars = s.char_indices().peekable();
let mut end = None;
while let Some((i, c)) = chars.next() {
if c == '/' && chars.peek().map(|(_, c2)| *c2) == Some('*') {
chars.next();
depth += 1;
} else if c == '*' && chars.peek().map(|(_, c2)| *c2) == Some('/') {
chars.next();
depth -= 1;
if depth == 0 {
end = Some(i + 2);
break;
}
}
}
match end {
Some(pos) => s = &s[pos..],
None => return "", }
} else {
break;
}
}
s
}
impl Drop for Engine {
fn drop(&mut self) {
if self.daemon_endpoint.is_some() {
let db_name = self.primary_db_name();
let detach = format!("DETACH DATABASE \"{db_name}\"");
let _ = self.connection.execute_command(&detach);
}
if let Some(parent) = self.ephemeral_path.parent() {
let _ = std::fs::remove_dir_all(parent);
}
}
}
fn probe_endpoint_alive(endpoint: &str) -> bool {
use std::net::ToSocketAddrs;
const PROBE_TIMEOUT: std::time::Duration = std::time::Duration::from_millis(300);
match endpoint.to_socket_addrs() {
Ok(mut addrs) => {
addrs.any(|addr| std::net::TcpStream::connect_timeout(&addr, PROBE_TIMEOUT).is_ok())
}
Err(_) => false,
}
}
fn bootstrap_public_schema(connection: &Connection) -> Result<(), McpError> {
connection
.execute_command("CREATE SCHEMA IF NOT EXISTS public")
.map(|_| ())
.map_err(|e| {
McpError::new(
ErrorCode::InternalError,
format!("Failed to bootstrap public schema: {e}"),
)
})
}
fn shellexpand_tilde(path: &str) -> String {
let rest = if let Some(r) = path.strip_prefix("~/") {
Some(r)
} else if cfg!(windows) {
path.strip_prefix("~\\")
} else {
None
};
let Some(rest) = rest else {
return path.to_string();
};
let Some(home) = home_dir() else {
return path.to_string();
};
let sep = std::path::MAIN_SEPARATOR;
format!("{}{sep}{rest}", home.to_string_lossy())
}
fn home_dir() -> Option<PathBuf> {
if cfg!(windows) {
if let Some(profile) = std::env::var_os("USERPROFILE") {
if !profile.is_empty() {
return Some(PathBuf::from(profile));
}
}
let drive = std::env::var_os("HOMEDRIVE")?;
let rel = std::env::var_os("HOMEPATH")?;
let mut combined = PathBuf::from(drive);
combined.push(PathBuf::from(rel));
Some(combined)
} else {
std::env::var_os("HOME").map(PathBuf::from)
}
}
#[cfg(test)]
mod statement_helper_tests {
use super::*;
#[test]
fn classify_statement_recognizes_each_kind() {
assert_eq!(classify_statement("SELECT 1"), StatementKind::ReadOnly);
assert_eq!(
classify_statement("with x as (..) select * from x"),
StatementKind::ReadOnly
);
assert_eq!(
classify_statement("CREATE TABLE t (i INT)"),
StatementKind::Ddl
);
assert_eq!(classify_statement("drop table t"), StatementKind::Ddl);
assert_eq!(
classify_statement("INSERT INTO t VALUES (1)"),
StatementKind::Dml
);
assert_eq!(classify_statement("update t set i = 2"), StatementKind::Dml);
assert_eq!(classify_statement("delete from t"), StatementKind::Dml);
assert_eq!(classify_statement(""), StatementKind::Other);
}
#[test]
fn classify_statement_recognizes_transaction_control() {
for kw in [
"BEGIN",
"Begin transaction",
"START TRANSACTION",
"COMMIT",
"Commit work",
"END",
"ROLLBACK",
"Rollback to savepoint sp1",
"ABORT",
"SAVEPOINT sp1",
"RELEASE SAVEPOINT sp1",
] {
assert_eq!(
classify_statement(kw),
StatementKind::TransactionControl,
"expected TransactionControl for `{kw}`"
);
}
}
#[test]
fn classify_statement_strips_comments() {
assert_eq!(
classify_statement("/* harmless */ DROP TABLE t"),
StatementKind::Ddl
);
assert_eq!(
classify_statement("-- pretend to be readonly\nINSERT INTO t VALUES (1)"),
StatementKind::Dml
);
}
}