1use std::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(
311 CliError::Command("Must specify --applied or --rolled-back".to_string()).into(),
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: &PathBuf) -> 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: &PathBuf) -> 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() {
460 if !is_migration_applied(&path)? {
461 pending.push(path);
462 }
463 }
464 }
465
466 pending.sort();
467 Ok(pending)
468}
469
470fn is_migration_applied(migration_path: &PathBuf) -> CliResult<bool> {
471 let marker = migration_path.join(".applied");
474 Ok(marker.exists())
475}
476
477fn create_migration(
478 migrations_dir: &PathBuf,
479 name: &str,
480 schema: &prax_schema::ast::Schema,
481) -> CliResult<PathBuf> {
482 let timestamp = chrono::Utc::now().format("%Y%m%d%H%M%S");
484 let migration_name = format!("{}_{}", timestamp, name);
485 let migration_path = migrations_dir.join(&migration_name);
486
487 std::fs::create_dir_all(&migration_path)?;
488
489 let sql = generate_schema_diff(schema)?;
491
492 let sql_path = migration_path.join("migration.sql");
494 std::fs::write(&sql_path, &sql)?;
495
496 Ok(migration_path)
497}
498
499fn generate_schema_diff(schema: &prax_schema::ast::Schema) -> CliResult<String> {
500 use prax_schema::ast::{FieldType, ScalarType};
501
502 let mut sql = String::new();
503
504 sql.push_str("-- Migration generated by Prax\n\n");
505
506 if !schema.enums.is_empty() {
508 sql.push_str("-- Enum types\n");
509 for enum_def in schema.enums.values() {
510 let enum_name = enum_def
511 .attributes
512 .iter()
513 .find(|a| a.is("map"))
514 .and_then(|a: &prax_schema::ast::Attribute| a.first_arg())
515 .and_then(|v: &prax_schema::ast::AttributeValue| v.as_string())
516 .map(|s| s.to_string())
517 .unwrap_or_else(|| to_snake_case(enum_def.name()));
518
519 sql.push_str(&format!(
520 "DO $$ BEGIN\n CREATE TYPE \"{}\" AS ENUM (",
521 enum_name
522 ));
523
524 let variants: Vec<String> = enum_def
525 .variants
526 .iter()
527 .map(|v| format!("'{}'", v.name()))
528 .collect();
529
530 sql.push_str(&variants.join(", "));
531 sql.push_str(");\nEXCEPTION\n WHEN duplicate_object THEN null;\nEND $$;\n\n");
532 }
533 sql.push_str("\n");
534 }
535
536 sql.push_str("-- Tables\n");
538 for model in schema.models.values() {
539 let table_name = model.table_name();
540
541 sql.push_str(&format!(
542 "CREATE TABLE IF NOT EXISTS \"{}\" (\n",
543 table_name
544 ));
545
546 let mut columns = Vec::new();
547 let mut primary_keys = Vec::new();
548
549 for field in model.fields.values() {
550 if field.is_relation() {
551 continue;
552 }
553
554 let column_name = field
555 .get_attribute("map")
556 .and_then(|a| a.first_arg())
557 .and_then(|v| v.as_string())
558 .map(|s| s.to_string())
559 .unwrap_or_else(|| to_snake_case(field.name()));
560
561 let sql_type = field_type_to_sql(&field.field_type);
562 let mut column_def = format!(" \"{}\" {}", column_name, sql_type);
563
564 if field.is_id() {
566 primary_keys.push(column_name.clone());
567 }
568
569 if field.has_attribute("auto") || field.has_attribute("autoincrement") {
570 column_def = format!(" \"{}\" SERIAL", column_name);
572 }
573
574 if field.has_attribute("unique") {
575 column_def.push_str(" UNIQUE");
576 }
577
578 if !field.is_optional() && !field.is_id() {
579 column_def.push_str(" NOT NULL");
580 }
581
582 if let Some(default_attr) = field.get_attribute("default") {
584 if let Some(value) = default_attr.first_arg() {
585 let value_str = format_attribute_value(value);
586 column_def.push_str(&format!(
587 " DEFAULT {}",
588 sql_default_value(&value_str, &field.field_type)
589 ));
590 }
591 }
592
593 columns.push(column_def);
594 }
595
596 sql.push_str(&columns.join(",\n"));
597
598 if !primary_keys.is_empty() {
599 sql.push_str(",\n");
600 sql.push_str(&format!(
601 " PRIMARY KEY (\"{}\")",
602 primary_keys.join("\", \"")
603 ));
604 }
605
606 sql.push_str("\n);\n\n");
607 }
608
609 return Ok(sql);
610
611 fn field_type_to_sql(field_type: &FieldType) -> String {
612 match field_type {
613 FieldType::Scalar(scalar) => match scalar {
614 ScalarType::Int => "INTEGER".to_string(),
615 ScalarType::BigInt => "BIGINT".to_string(),
616 ScalarType::Float => "DOUBLE PRECISION".to_string(),
617 ScalarType::String => "TEXT".to_string(),
618 ScalarType::Boolean => "BOOLEAN".to_string(),
619 ScalarType::DateTime => "TIMESTAMP WITH TIME ZONE".to_string(),
620 ScalarType::Date => "DATE".to_string(),
621 ScalarType::Time => "TIME".to_string(),
622 ScalarType::Json => "JSONB".to_string(),
623 ScalarType::Bytes => "BYTEA".to_string(),
624 ScalarType::Decimal => "DECIMAL".to_string(),
625 ScalarType::Uuid => "UUID".to_string(),
626 ScalarType::Cuid | ScalarType::Cuid2 | ScalarType::NanoId | ScalarType::Ulid => {
627 "TEXT".to_string()
628 }
629 ScalarType::Vector(dim) => match dim {
630 Some(d) => format!("vector({})", d),
631 None => "vector".to_string(),
632 },
633 ScalarType::HalfVector(dim) => match dim {
634 Some(d) => format!("halfvec({})", d),
635 None => "halfvec".to_string(),
636 },
637 ScalarType::SparseVector(dim) => match dim {
638 Some(d) => format!("sparsevec({})", d),
639 None => "sparsevec".to_string(),
640 },
641 ScalarType::Bit(dim) => match dim {
642 Some(d) => format!("bit({})", d),
643 None => "bit".to_string(),
644 },
645 },
646 FieldType::Enum(name) => format!("\"{}\"", to_snake_case(name)),
647 _ => "TEXT".to_string(),
648 }
649 }
650}
651
652async fn apply_migration(migration_path: &PathBuf, _config: &Config) -> CliResult<()> {
653 let sql_path = migration_path.join("migration.sql");
654
655 if !sql_path.exists() {
656 return Err(CliError::Migration(format!(
657 "Migration file not found: {}",
658 sql_path.display()
659 )));
660 }
661
662 let _sql = std::fs::read_to_string(&sql_path)?;
663
664 let marker = migration_path.join(".applied");
669 std::fs::write(&marker, chrono::Utc::now().to_rfc3339())?;
670
671 Ok(())
672}
673
674fn sql_default_value(value: &str, field_type: &prax_schema::ast::FieldType) -> String {
675 if matches!(field_type, prax_schema::ast::FieldType::Enum(_)) {
677 return format!("'{}'", value);
678 }
679
680 match value.to_lowercase().as_str() {
681 "now()" => "CURRENT_TIMESTAMP".to_string(),
682 "uuid()" => "gen_random_uuid()".to_string(),
683 "cuid()" | "cuid2()" | "nanoid()" | "ulid()" => {
684 "''".to_string()
686 }
687 "true" => "TRUE".to_string(),
688 "false" => "FALSE".to_string(),
689 _ => value.to_string(),
690 }
691}
692
693fn to_snake_case(name: &str) -> String {
694 let mut result = String::new();
695 for (i, c) in name.chars().enumerate() {
696 if c.is_uppercase() {
697 if i > 0 {
698 result.push('_');
699 }
700 result.push(c.to_lowercase().next().unwrap());
701 } else {
702 result.push(c);
703 }
704 }
705 result
706}
707
708fn format_attribute_value(value: &prax_schema::ast::AttributeValue) -> String {
709 use prax_schema::ast::AttributeValue;
710
711 match value {
712 AttributeValue::String(s) => format!("\"{}\"", s),
713 AttributeValue::Int(i) => i.to_string(),
714 AttributeValue::Float(f) => f.to_string(),
715 AttributeValue::Boolean(b) => b.to_string(),
716 AttributeValue::Ident(id) => id.to_string(),
717 AttributeValue::Function(name, args) => {
718 if args.is_empty() {
719 format!("{}()", name)
720 } else {
721 let arg_strs: Vec<String> = args.iter().map(format_attribute_value).collect();
722 format!("{}({})", name, arg_strs.join(", "))
723 }
724 }
725 AttributeValue::Array(items) => {
726 let item_strs: Vec<String> = items.iter().map(format_attribute_value).collect();
727 format!("[{}]", item_strs.join(", "))
728 }
729 AttributeValue::FieldRef(field) => field.to_string(),
730 AttributeValue::FieldRefList(fields) => {
731 format!(
732 "[{}]",
733 fields
734 .iter()
735 .map(|f| f.to_string())
736 .collect::<Vec<_>>()
737 .join(", ")
738 )
739 }
740 }
741}