use anyhow::{Context, Result};
use clap::Parser;
use serde_json::Value;
use std::fs;
use std::path::{Path, PathBuf};
use std::process::Command;
#[derive(Parser, Debug)]
#[command(version, about, long_about = None)]
struct Args {
#[arg(short, long)]
from: PathBuf,
#[arg(short, long)]
to: PathBuf,
#[arg(long, default_value_t = false)]
skip_extract: bool,
}
fn main() -> Result<()> {
let args = Args::parse();
println!("Starting migration from {:?} to {:?}", args.from, args.to);
let temp_dir = if args.skip_extract {
args.from.clone()
} else {
let dir = args.to.join(".migration_tmp");
fs::create_dir_all(&dir)?;
dir
};
if !args.skip_extract {
let python_script = Path::new(env!("CARGO_MANIFEST_DIR")).join("src").join("export_cpp.py");
println!("1. Extracting data and schema from C++ Akar (via Python)...");
let status = Command::new("python")
.arg(&python_script)
.arg("--db_path")
.arg(&args.from)
.arg("--out_dir")
.arg(&temp_dir)
.status()
.context(
"Failed to execute python extraction script. Make sure Python and akar/ladybug packages are installed.",
)?;
if !status.success() {
anyhow::bail!("Python extraction script failed");
}
} else {
println!("1. Skipping extraction, reading directly from {:?}", temp_dir);
}
println!("2. Connecting to Rust Akar Database...");
let rust_db = std::sync::Arc::new(
akar_main::Database::new(&args.to, akar_main::SystemConfig::default())
.map_err(|e| anyhow::anyhow!("DB Init Error: {}", e))?,
);
let rust_conn = akar_main::Connection::new(&rust_db);
let schema_file = temp_dir.join("schema.json");
let schema_json = fs::read_to_string(&schema_file)?;
let schema: Value = serde_json::from_str(&schema_json)?;
let tables = schema["tables"]
.as_array()
.context("Invalid schema: tables is not an array")?;
let connections = schema["connections"]
.as_array()
.context("Invalid schema: connections is not an array")?;
println!("3. Reconstructing DDL and loading data...");
for table in tables {
let table_type = table["type"].as_str().unwrap_or("");
if table_type != "NODE" {
continue;
}
let table_name = table["name"].as_str().unwrap();
if rust_db.get_table_id(table_name).is_some() {
println!("Skipping {} (already exists)", table_name);
continue;
}
let properties = table["properties"].as_array().unwrap();
let mut columns = Vec::new();
let mut primary_key = String::new();
for prop in properties {
let col_name = prop["name"].as_str().unwrap();
let col_type = prop["type"].as_str().unwrap();
let is_pk = prop["is_primary_key"].as_bool().unwrap_or(false);
columns.push(format!("{} {}", col_name, col_type));
if is_pk {
primary_key = col_name.to_string();
}
}
let ddl = if !primary_key.is_empty() {
format!(
"CREATE NODE TABLE {} ({}, PRIMARY KEY({}))",
table_name,
columns.join(", "),
primary_key
)
} else {
format!("CREATE NODE TABLE {} ({})", table_name, columns.join(", "))
};
println!("Executing: {}", ddl);
rust_conn.query(&ddl).map_err(|e| anyhow::anyhow!("DDL Error: {}", e))?;
let parquet_path = temp_dir.join(format!("{}.parquet", table_name));
let parquet_path_str = parquet_path.to_str().unwrap().replace("\\", "/");
let import_query = format!("COPY {} FROM '{}'", table_name, parquet_path_str);
println!("Importing data to {}...", table_name);
rust_conn
.query(&import_query)
.map_err(|e| anyhow::anyhow!("COPY Error: {}", e))?;
}
for table in tables {
let table_type = table["type"].as_str().unwrap_or("");
if table_type != "REL" {
continue;
}
let table_name = table["name"].as_str().unwrap();
if rust_db.get_table_id(table_name).is_some() {
println!("Skipping {} (already exists)", table_name);
continue;
}
let mut from_table = "UNKNOWN";
let mut to_table = "UNKNOWN";
for conn_info in connections {
if conn_info["rel"].as_str() == Some(table_name) {
from_table = conn_info["src"].as_str().unwrap();
to_table = conn_info["dst"].as_str().unwrap();
break;
}
}
let properties = table["properties"].as_array().unwrap();
let mut columns = Vec::new();
for prop in properties {
let col_name = prop["name"].as_str().unwrap();
let col_type = prop["type"].as_str().unwrap();
columns.push(format!("{} {}", col_name, col_type));
}
let props_str = if columns.is_empty() {
String::new()
} else {
format!(", {}", columns.join(", "))
};
let ddl = format!(
"CREATE REL TABLE {} (FROM {} TO {}{})",
table_name, from_table, to_table, props_str
);
println!("Executing: {}", ddl);
rust_conn.query(&ddl).map_err(|e| anyhow::anyhow!("DDL Error: {}", e))?;
let parquet_path = temp_dir.join(format!("{}.parquet", table_name));
let parquet_path_str = parquet_path.to_str().unwrap().replace("\\", "/");
let import_query = format!("COPY {} FROM '{}'", table_name, parquet_path_str);
println!("Importing data to {}...", table_name);
rust_conn
.query(&import_query)
.map_err(|e| anyhow::anyhow!("COPY Error: {}", e))?;
}
if !args.skip_extract {
println!("4. Cleaning up temporary files...");
let _ = fs::remove_dir_all(&temp_dir);
}
println!("Migration complete!");
Ok(())
}