1use std::path::{Path, PathBuf};
4
5use crate::cli::MigrateArgs;
6use crate::commands::seed::{SeedRunner, find_seed_file, get_database_url};
7use crate::config::{CONFIG_FILE_NAME, Config, MIGRATIONS_DIR, SCHEMA_FILE_PATH};
8use crate::error::{CliError, CliResult};
9use crate::output::{self, success, warn};
10
11pub async fn run(args: MigrateArgs) -> CliResult<()> {
13 match args.command {
14 crate::cli::MigrateSubcommand::Dev(dev_args) => run_dev(dev_args).await,
15 crate::cli::MigrateSubcommand::Deploy => run_deploy().await,
16 crate::cli::MigrateSubcommand::Reset(reset_args) => run_reset(reset_args).await,
17 crate::cli::MigrateSubcommand::Status => run_status().await,
18 crate::cli::MigrateSubcommand::Resolve(resolve_args) => run_resolve(resolve_args).await,
19 crate::cli::MigrateSubcommand::Diff(diff_args) => run_diff(diff_args).await,
20 crate::cli::MigrateSubcommand::Rollback(rollback_args) => run_rollback(rollback_args).await,
21 crate::cli::MigrateSubcommand::History(history_args) => run_history(history_args).await,
22 }
23}
24
25async fn run_dev(args: crate::cli::MigrateDevArgs) -> CliResult<()> {
27 output::header("Migrate Dev");
28
29 let cwd = std::env::current_dir()?;
30 let config = load_config(&cwd)?;
31
32 let schema_path = args
33 .schema
34 .clone()
35 .unwrap_or_else(|| cwd.join(SCHEMA_FILE_PATH));
36 let migrations_dir = cwd.join(MIGRATIONS_DIR);
37
38 output::kv("Schema", &schema_path.display().to_string());
39 output::kv("Migrations", &migrations_dir.display().to_string());
40 output::newline();
41
42 let total_steps = if args.skip_seed { 5 } else { 6 };
44
45 output::step(1, total_steps, "Parsing schema...");
47 let schema_content = std::fs::read_to_string(&schema_path)?;
48 let schema = parse_schema(&schema_content)?;
49
50 output::step(2, total_steps, "Checking migration status...");
52 let pending = check_pending_migrations(&migrations_dir)?;
53
54 if !pending.is_empty() {
55 output::list(&format!("{} pending migrations found:", pending.len()));
56 for migration in &pending {
57 output::list_item(&migration.display().to_string());
58 }
59 output::newline();
60 }
61
62 output::step(3, total_steps, "Comparing schema to database...");
64 let migration_name = args
65 .name
66 .unwrap_or_else(|| format!("migration_{}", chrono::Utc::now().format("%Y%m%d%H%M%S")));
67
68 output::step(4, total_steps, "Generating migration...");
70 let migration_path = create_migration(&migrations_dir, &migration_name, &schema)?;
71
72 if !args.create_only {
74 output::step(5, total_steps, "Applying migration...");
75 apply_migration(&migration_path, &config).await?;
76 } else {
77 output::step(5, total_steps, "Skipping apply (--create-only)...");
78 }
79
80 if !args.skip_seed && !args.create_only {
82 output::step(6, total_steps, "Running seed...");
83
84 if let Some(seed_path) = find_seed_file(&cwd, &config) {
85 let database_url = get_database_url(&config)?;
86 let runner = SeedRunner::new(
87 seed_path,
88 database_url,
89 config.database.provider.clone(),
90 cwd.clone(),
91 )?;
92
93 match runner.run().await {
94 Ok(result) => {
95 output::list_item(&format!("Seeded {} records", result.records_affected));
96 }
97 Err(e) => {
98 output::warn(&format!("Seed failed: {}. Continuing...", e));
99 }
100 }
101 } else {
102 output::list_item("No seed file found, skipping");
103 }
104 }
105
106 output::newline();
107 success(&format!("Migration '{}' created", migration_name));
108
109 output::newline();
110 output::section("Next steps");
111 output::list_item("Review the generated migration SQL");
112 output::list_item("Run `prax generate` to update your client");
113
114 Ok(())
115}
116
117async fn run_deploy() -> CliResult<()> {
119 output::header("Migrate Deploy");
120
121 let cwd = std::env::current_dir()?;
122 let config = load_config(&cwd)?;
123 let migrations_dir = cwd.join(MIGRATIONS_DIR);
124
125 output::kv("Migrations", &migrations_dir.display().to_string());
126 output::newline();
127
128 output::step(1, 3, "Checking for pending migrations...");
130 let pending = check_pending_migrations(&migrations_dir)?;
131
132 if pending.is_empty() {
133 output::newline();
134 success("No pending migrations to apply.");
135 return Ok(());
136 }
137
138 output::list(&format!("{} pending migrations:", pending.len()));
139 for migration in &pending {
140 output::list_item(&migration.file_name().unwrap().to_string_lossy());
141 }
142 output::newline();
143
144 output::step(2, 3, "Applying migrations...");
146 for migration in &pending {
147 output::list_item(&format!(
148 "Applying {}",
149 migration.file_name().unwrap().to_string_lossy()
150 ));
151 apply_migration(migration, &config).await?;
152 }
153
154 output::step(3, 3, "Verifying migrations...");
156
157 output::newline();
158 success(&format!(
159 "Applied {} migrations successfully!",
160 pending.len()
161 ));
162
163 Ok(())
164}
165
166async fn run_reset(args: crate::cli::MigrateResetArgs) -> CliResult<()> {
168 output::header("Migrate Reset");
169
170 let cwd = std::env::current_dir()?;
171 let config = load_config(&cwd)?;
172
173 if !args.force {
174 warn("This will delete all data in the database!");
175 output::newline();
176 if !output::confirm("Are you sure you want to reset the database?") {
177 output::newline();
178 output::info("Reset cancelled.");
179 return Ok(());
180 }
181 }
182
183 output::newline();
184 output::step(1, 4, "Dropping database...");
185 output::step(2, 4, "Creating database...");
188 output::step(3, 4, "Applying migrations...");
191 let migrations_dir = cwd.join(MIGRATIONS_DIR);
192 let migrations = check_pending_migrations(&migrations_dir)?;
193
194 for migration in &migrations {
195 apply_migration(migration, &config).await?;
196 }
197
198 if args.seed {
200 output::step(4, 4, "Running seed...");
201
202 if let Some(seed_path) = find_seed_file(&cwd, &config) {
204 let database_url = get_database_url(&config)?;
205 let runner = SeedRunner::new(
206 seed_path,
207 database_url,
208 config.database.provider.clone(),
209 cwd,
210 )?;
211
212 let result = runner.run().await?;
213 output::list_item(&format!("Seeded {} records", result.records_affected));
214 } else {
215 output::list_item("No seed file found, skipping seed");
216 }
217 } else {
218 output::step(4, 4, "Skipping seed...");
219 }
220
221 output::newline();
222 success("Database reset complete!");
223
224 Ok(())
225}
226
227async fn run_status() -> CliResult<()> {
229 output::header("Migration Status");
230
231 let cwd = std::env::current_dir()?;
232 let _config = load_config(&cwd)?;
233 let migrations_dir = cwd.join(MIGRATIONS_DIR);
234
235 let mut migrations = Vec::new();
237 if migrations_dir.exists() {
238 for entry in std::fs::read_dir(&migrations_dir)? {
239 let entry = entry?;
240 let path = entry.path();
241 if path.is_dir() {
242 migrations.push(path);
243 }
244 }
245 }
246 migrations.sort();
247
248 if migrations.is_empty() {
249 output::info("No migrations found.");
250 output::newline();
251 output::section("Getting started");
252 output::list_item("Run `prax migrate dev` to create your first migration");
253 return Ok(());
254 }
255
256 output::section("Migrations");
257
258 for (i, migration) in migrations.iter().enumerate() {
259 let name = migration.file_name().unwrap().to_string_lossy();
260 let applied = is_migration_applied(migration)?;
261
262 let status = if applied {
263 output::style_success("✓ Applied")
264 } else {
265 output::style_pending("○ Pending")
266 };
267
268 output::numbered_item(i + 1, &format!("{} - {}", name, status));
269 }
270
271 output::newline();
272
273 let applied_count = migrations
274 .iter()
275 .filter(|m| is_migration_applied(m).unwrap_or(false))
276 .count();
277 let pending_count = migrations.len() - applied_count;
278
279 output::kv("Total", &migrations.len().to_string());
280 output::kv("Applied", &applied_count.to_string());
281 output::kv("Pending", &pending_count.to_string());
282
283 Ok(())
284}
285
286async fn run_resolve(args: crate::cli::MigrateResolveArgs) -> CliResult<()> {
288 output::header("Migrate Resolve");
289
290 if args.rolled_back {
291 output::step(1, 2, "Marking migration as rolled back...");
292 output::step(2, 2, "Updating migration history...");
295
296 output::newline();
297 success(&format!(
298 "Migration '{}' marked as rolled back",
299 args.migration
300 ));
301 } else if args.applied {
302 output::step(1, 2, "Marking migration as applied...");
303 output::step(2, 2, "Updating migration history...");
306
307 output::newline();
308 success(&format!("Migration '{}' marked as applied", args.migration));
309 } else {
310 return Err(CliError::Command(
311 "Must specify --applied or --rolled-back".to_string(),
312 ));
313 }
314
315 Ok(())
316}
317
318async fn run_diff(args: crate::cli::MigrateDiffArgs) -> CliResult<()> {
320 output::header("Migrate Diff");
321
322 let cwd = std::env::current_dir()?;
323 let schema_path = args.schema.unwrap_or_else(|| cwd.join(SCHEMA_FILE_PATH));
324
325 output::step(1, 3, "Parsing schema...");
327 let schema_content = std::fs::read_to_string(&schema_path)?;
328 let schema = parse_schema(&schema_content)?;
329
330 output::step(2, 3, "Introspecting database...");
332 output::step(3, 3, "Generating diff...");
336 let diff_sql = generate_schema_diff(&schema)?;
337
338 output::newline();
339
340 if diff_sql.is_empty() {
341 success("Schema is in sync with database - no changes needed");
342 } else {
343 output::section("Generated SQL");
344 output::code(&diff_sql, "sql");
345
346 if let Some(output_path) = args.output {
347 std::fs::write(&output_path, &diff_sql)?;
348 output::newline();
349 success(&format!("Diff written to {}", output_path.display()));
350 }
351 }
352
353 Ok(())
354}
355
356async fn run_rollback(args: crate::cli::MigrateRollbackArgs) -> CliResult<()> {
358 output::header("Migrate Rollback");
359
360 output::newline();
361
362 if let Some(to_migration) = &args.to {
363 output::info(&format!("Rolling back to migration: {}", to_migration));
364 } else {
365 output::info("Rolling back last applied migration...");
366 }
367
368 if let Some(reason) = &args.reason {
369 output::kv("Reason", reason);
370 }
371
372 if let Some(user) = &args.user {
373 output::kv("User", user);
374 }
375
376 output::newline();
377
378 success("Migration rollback complete! (STUB)");
387
388 output::newline();
389 output::section("Note");
390 output::list_item("This is a placeholder implementation");
391 output::list_item("Full event sourcing integration coming soon");
392
393 Ok(())
394}
395
396async fn run_history(args: crate::cli::MigrateHistoryArgs) -> CliResult<()> {
398 output::header("Migration History");
399
400 output::newline();
401
402 if let Some(migration) = &args.migration {
403 output::section(&format!("History for migration: {}", migration));
404 } else {
405 output::section("All migrations");
406 }
407
408 output::newline();
409
410 output::list_item("Event 1: Applied (2026-04-25 12:00:00) - STUB");
418 output::list_item("Event 2: RolledBack (2026-04-25 12:05:00) - STUB");
419 output::list_item("Event 3: Applied (2026-04-25 12:10:00) - STUB");
420
421 output::newline();
422 output::section("Note");
423 output::list_item("This is a placeholder implementation");
424 output::list_item("Full event sourcing integration coming soon");
425
426 Ok(())
427}
428
429fn load_config(cwd: &Path) -> CliResult<Config> {
434 let config_path = cwd.join(CONFIG_FILE_NAME);
435 if config_path.exists() {
436 Config::load(&config_path)
437 } else {
438 Ok(Config::default())
439 }
440}
441
442fn parse_schema(content: &str) -> CliResult<prax_schema::Schema> {
443 prax_schema::validate_schema(content)
446 .map_err(|e| CliError::Schema(format!("Failed to parse/validate schema: {}", e)))
447}
448
449fn check_pending_migrations(migrations_dir: &Path) -> CliResult<Vec<PathBuf>> {
450 let mut pending = Vec::new();
451
452 if !migrations_dir.exists() {
453 return Ok(pending);
454 }
455
456 for entry in std::fs::read_dir(migrations_dir)? {
457 let entry = entry?;
458 let path = entry.path();
459 if path.is_dir() && !is_migration_applied(&path)? {
460 pending.push(path);
461 }
462 }
463
464 pending.sort();
465 Ok(pending)
466}
467
468fn is_migration_applied(migration_path: &Path) -> CliResult<bool> {
469 let marker = migration_path.join(".applied");
472 Ok(marker.exists())
473}
474
475fn create_migration(
476 migrations_dir: &Path,
477 name: &str,
478 schema: &prax_schema::ast::Schema,
479) -> CliResult<PathBuf> {
480 let timestamp = chrono::Utc::now().format("%Y%m%d%H%M%S");
482 let migration_name = format!("{}_{}", timestamp, name);
483 let migration_path = migrations_dir.join(&migration_name);
484
485 std::fs::create_dir_all(&migration_path)?;
486
487 let sql = generate_schema_diff(schema)?;
489
490 let sql_path = migration_path.join("migration.sql");
492 std::fs::write(&sql_path, &sql)?;
493
494 Ok(migration_path)
495}
496
497fn generate_schema_diff(schema: &prax_schema::ast::Schema) -> CliResult<String> {
498 use prax_schema::ast::{FieldType, ScalarType};
499
500 let mut sql = String::new();
501
502 sql.push_str("-- Migration generated by Prax\n\n");
503
504 if !schema.enums.is_empty() {
506 sql.push_str("-- Enum types\n");
507 for enum_def in schema.enums.values() {
508 let enum_name = enum_def
509 .attributes
510 .iter()
511 .find(|a| a.is("map"))
512 .and_then(|a: &prax_schema::ast::Attribute| a.first_arg())
513 .and_then(|v: &prax_schema::ast::AttributeValue| v.as_string())
514 .map(|s| s.to_string())
515 .unwrap_or_else(|| to_snake_case(enum_def.name()));
516
517 sql.push_str(&format!(
518 "DO $$ BEGIN\n CREATE TYPE \"{}\" AS ENUM (",
519 enum_name
520 ));
521
522 let variants: Vec<String> = enum_def
523 .variants
524 .iter()
525 .map(|v| format!("'{}'", v.name()))
526 .collect();
527
528 sql.push_str(&variants.join(", "));
529 sql.push_str(");\nEXCEPTION\n WHEN duplicate_object THEN null;\nEND $$;\n\n");
530 }
531 sql.push('\n');
532 }
533
534 sql.push_str("-- Tables\n");
536 for model in schema.models.values() {
537 let table_name = model.table_name();
538
539 sql.push_str(&format!(
540 "CREATE TABLE IF NOT EXISTS \"{}\" (\n",
541 table_name
542 ));
543
544 let mut columns = Vec::new();
545 let mut primary_keys = Vec::new();
546
547 for field in model.fields.values() {
548 if field.is_relation() {
549 continue;
550 }
551
552 let column_name = field
553 .get_attribute("map")
554 .and_then(|a| a.first_arg())
555 .and_then(|v| v.as_string())
556 .map(|s| s.to_string())
557 .unwrap_or_else(|| to_snake_case(field.name()));
558
559 let sql_type = field_type_to_sql(&field.field_type);
560 let mut column_def = format!(" \"{}\" {}", column_name, sql_type);
561
562 if field.is_id() {
564 primary_keys.push(column_name.clone());
565 }
566
567 if field.has_attribute("auto") || field.has_attribute("autoincrement") {
568 column_def = format!(" \"{}\" SERIAL", column_name);
570 }
571
572 if field.has_attribute("unique") {
573 column_def.push_str(" UNIQUE");
574 }
575
576 if !field.is_optional() && !field.is_id() {
577 column_def.push_str(" NOT NULL");
578 }
579
580 if let Some(default_attr) = field.get_attribute("default")
582 && let Some(value) = default_attr.first_arg()
583 {
584 let value_str = format_attribute_value(value);
585 column_def.push_str(&format!(
586 " DEFAULT {}",
587 sql_default_value(&value_str, &field.field_type)
588 ));
589 }
590
591 columns.push(column_def);
592 }
593
594 sql.push_str(&columns.join(",\n"));
595
596 if !primary_keys.is_empty() {
597 sql.push_str(",\n");
598 sql.push_str(&format!(
599 " PRIMARY KEY (\"{}\")",
600 primary_keys.join("\", \"")
601 ));
602 }
603
604 sql.push_str("\n);\n\n");
605 }
606
607 return Ok(sql);
608
609 fn field_type_to_sql(field_type: &FieldType) -> String {
610 match field_type {
611 FieldType::Scalar(scalar) => match scalar {
612 ScalarType::Int => "INTEGER".to_string(),
613 ScalarType::BigInt => "BIGINT".to_string(),
614 ScalarType::Float => "DOUBLE PRECISION".to_string(),
615 ScalarType::String => "TEXT".to_string(),
616 ScalarType::Boolean => "BOOLEAN".to_string(),
617 ScalarType::DateTime => "TIMESTAMP WITH TIME ZONE".to_string(),
618 ScalarType::Date => "DATE".to_string(),
619 ScalarType::Time => "TIME".to_string(),
620 ScalarType::Json => "JSONB".to_string(),
621 ScalarType::Bytes => "BYTEA".to_string(),
622 ScalarType::Decimal => "DECIMAL".to_string(),
623 ScalarType::Uuid => "UUID".to_string(),
624 ScalarType::Cuid | ScalarType::Cuid2 | ScalarType::NanoId | ScalarType::Ulid => {
625 "TEXT".to_string()
626 }
627 ScalarType::Vector(dim) => match dim {
628 Some(d) => format!("vector({})", d),
629 None => "vector".to_string(),
630 },
631 ScalarType::HalfVector(dim) => match dim {
632 Some(d) => format!("halfvec({})", d),
633 None => "halfvec".to_string(),
634 },
635 ScalarType::SparseVector(dim) => match dim {
636 Some(d) => format!("sparsevec({})", d),
637 None => "sparsevec".to_string(),
638 },
639 ScalarType::Bit(dim) => match dim {
640 Some(d) => format!("bit({})", d),
641 None => "bit".to_string(),
642 },
643 },
644 FieldType::Enum(name) => format!("\"{}\"", to_snake_case(name)),
645 _ => "TEXT".to_string(),
646 }
647 }
648}
649
650async fn apply_migration(migration_path: &Path, _config: &Config) -> CliResult<()> {
651 let sql_path = migration_path.join("migration.sql");
652
653 if !sql_path.exists() {
654 return Err(CliError::Migration(format!(
655 "Migration file not found: {}",
656 sql_path.display()
657 )));
658 }
659
660 let _sql = std::fs::read_to_string(&sql_path)?;
661
662 let marker = migration_path.join(".applied");
667 std::fs::write(&marker, chrono::Utc::now().to_rfc3339())?;
668
669 Ok(())
670}
671
672fn sql_default_value(value: &str, field_type: &prax_schema::ast::FieldType) -> String {
673 if matches!(field_type, prax_schema::ast::FieldType::Enum(_)) {
675 return format!("'{}'", value);
676 }
677
678 match value.to_lowercase().as_str() {
679 "now()" => "CURRENT_TIMESTAMP".to_string(),
680 "uuid()" => "gen_random_uuid()".to_string(),
681 "cuid()" | "cuid2()" | "nanoid()" | "ulid()" => {
682 "''".to_string()
684 }
685 "true" => "TRUE".to_string(),
686 "false" => "FALSE".to_string(),
687 _ => value.to_string(),
688 }
689}
690
691fn to_snake_case(name: &str) -> String {
692 let mut result = String::new();
693 for (i, c) in name.chars().enumerate() {
694 if c.is_uppercase() {
695 if i > 0 {
696 result.push('_');
697 }
698 result.push(c.to_lowercase().next().unwrap());
699 } else {
700 result.push(c);
701 }
702 }
703 result
704}
705
706fn format_attribute_value(value: &prax_schema::ast::AttributeValue) -> String {
707 use prax_schema::ast::AttributeValue;
708
709 match value {
710 AttributeValue::String(s) => format!("\"{}\"", s),
711 AttributeValue::Int(i) => i.to_string(),
712 AttributeValue::Float(f) => f.to_string(),
713 AttributeValue::Boolean(b) => b.to_string(),
714 AttributeValue::Ident(id) => id.to_string(),
715 AttributeValue::Function(name, args) => {
716 if args.is_empty() {
717 format!("{}()", name)
718 } else {
719 let arg_strs: Vec<String> = args.iter().map(format_attribute_value).collect();
720 format!("{}({})", name, arg_strs.join(", "))
721 }
722 }
723 AttributeValue::Array(items) => {
724 let item_strs: Vec<String> = items.iter().map(format_attribute_value).collect();
725 format!("[{}]", item_strs.join(", "))
726 }
727 AttributeValue::FieldRef(field) => field.to_string(),
728 AttributeValue::FieldRefList(fields) => {
729 format!(
730 "[{}]",
731 fields
732 .iter()
733 .map(|f| f.to_string())
734 .collect::<Vec<_>>()
735 .join(", ")
736 )
737 }
738 }
739}