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 migrations_dir = cwd.join(MIGRATIONS_DIR);
33
34 let display_path = args
35 .schema
36 .as_deref()
37 .map(|p| p.display().to_string())
38 .unwrap_or_else(|| SCHEMA_FILE_PATH.to_string());
39 output::kv("Schema", &display_path);
40 output::kv("Migrations", &migrations_dir.display().to_string());
41 output::newline();
42
43 let total_steps = if args.skip_seed { 5 } else { 6 };
45
46 output::step(1, total_steps, "Parsing schema...");
48 let loaded = crate::schema_loader::load_schema(args.schema.as_deref())?;
49 let schema = loaded.schema;
50
51 output::step(2, total_steps, "Checking migration status...");
53 let pending = check_pending_migrations(&migrations_dir)?;
54
55 if !pending.is_empty() {
56 output::list(&format!("{} pending migrations found:", pending.len()));
57 for migration in &pending {
58 output::list_item(&migration.display().to_string());
59 }
60 output::newline();
61 }
62
63 output::step(3, total_steps, "Comparing schema to database...");
65 let migration_name = args
66 .name
67 .unwrap_or_else(|| format!("migration_{}", chrono::Utc::now().format("%Y%m%d%H%M%S")));
68
69 output::step(4, total_steps, "Generating migration...");
71 let migration_path = create_migration(&migrations_dir, &migration_name, &schema)?;
72
73 if !args.create_only {
75 output::step(5, total_steps, "Applying migration...");
76 apply_migration(&migration_path, &config).await?;
77 } else {
78 output::step(5, total_steps, "Skipping apply (--create-only)...");
79 }
80
81 if !args.skip_seed && !args.create_only {
83 output::step(6, total_steps, "Running seed...");
84
85 if let Some(seed_path) = find_seed_file(&cwd, &config) {
86 let database_url = get_database_url(&config)?;
87 let runner = SeedRunner::new(
88 seed_path,
89 database_url,
90 config.database.provider.clone(),
91 cwd.clone(),
92 )?;
93
94 match runner.run().await {
95 Ok(result) => {
96 output::list_item(&format!("Seeded {} records", result.records_affected));
97 }
98 Err(e) => {
99 output::warn(&format!("Seed failed: {}. Continuing...", e));
100 }
101 }
102 } else {
103 output::list_item("No seed file found, skipping");
104 }
105 }
106
107 output::newline();
108 success(&format!("Migration '{}' created", migration_name));
109
110 output::newline();
111 output::section("Next steps");
112 output::list_item("Review the generated migration SQL");
113 output::list_item("Run `prax generate` to update your client");
114
115 Ok(())
116}
117
118async fn run_deploy() -> CliResult<()> {
120 output::header("Migrate Deploy");
121
122 let cwd = std::env::current_dir()?;
123 let config = load_config(&cwd)?;
124 let migrations_dir = cwd.join(MIGRATIONS_DIR);
125
126 output::kv("Migrations", &migrations_dir.display().to_string());
127 output::newline();
128
129 output::step(1, 3, "Checking for pending migrations...");
131 let pending = check_pending_migrations(&migrations_dir)?;
132
133 if pending.is_empty() {
134 output::newline();
135 success("No pending migrations to apply.");
136 return Ok(());
137 }
138
139 output::list(&format!("{} pending migrations:", pending.len()));
140 for migration in &pending {
141 output::list_item(&migration.file_name().unwrap().to_string_lossy());
142 }
143 output::newline();
144
145 output::step(2, 3, "Applying migrations...");
147 for migration in &pending {
148 output::list_item(&format!(
149 "Applying {}",
150 migration.file_name().unwrap().to_string_lossy()
151 ));
152 apply_migration(migration, &config).await?;
153 }
154
155 output::step(3, 3, "Verifying migrations...");
157
158 output::newline();
159 success(&format!(
160 "Applied {} migrations successfully!",
161 pending.len()
162 ));
163
164 Ok(())
165}
166
167async fn run_reset(args: crate::cli::MigrateResetArgs) -> CliResult<()> {
169 output::header("Migrate Reset");
170
171 let cwd = std::env::current_dir()?;
172 let config = load_config(&cwd)?;
173
174 if !args.force {
175 warn("This will delete all data in the database!");
176 output::newline();
177 if !output::confirm("Are you sure you want to reset the database?") {
178 output::newline();
179 output::info("Reset cancelled.");
180 return Ok(());
181 }
182 }
183
184 output::newline();
185 output::step(1, 4, "Dropping database...");
186 output::step(2, 4, "Creating database...");
189 output::step(3, 4, "Applying migrations...");
192 let migrations_dir = cwd.join(MIGRATIONS_DIR);
193 let migrations = check_pending_migrations(&migrations_dir)?;
194
195 for migration in &migrations {
196 apply_migration(migration, &config).await?;
197 }
198
199 if args.seed {
201 output::step(4, 4, "Running seed...");
202
203 if let Some(seed_path) = find_seed_file(&cwd, &config) {
205 let database_url = get_database_url(&config)?;
206 let runner = SeedRunner::new(
207 seed_path,
208 database_url,
209 config.database.provider.clone(),
210 cwd,
211 )?;
212
213 let result = runner.run().await?;
214 output::list_item(&format!("Seeded {} records", result.records_affected));
215 } else {
216 output::list_item("No seed file found, skipping seed");
217 }
218 } else {
219 output::step(4, 4, "Skipping seed...");
220 }
221
222 output::newline();
223 success("Database reset complete!");
224
225 Ok(())
226}
227
228async fn run_status() -> CliResult<()> {
230 output::header("Migration Status");
231
232 let cwd = std::env::current_dir()?;
233 let _config = load_config(&cwd)?;
234 let migrations_dir = cwd.join(MIGRATIONS_DIR);
235
236 let mut migrations = Vec::new();
238 if migrations_dir.exists() {
239 for entry in std::fs::read_dir(&migrations_dir)? {
240 let entry = entry?;
241 let path = entry.path();
242 if path.is_dir() {
243 migrations.push(path);
244 }
245 }
246 }
247 migrations.sort();
248
249 if migrations.is_empty() {
250 output::info("No migrations found.");
251 output::newline();
252 output::section("Getting started");
253 output::list_item("Run `prax migrate dev` to create your first migration");
254 return Ok(());
255 }
256
257 output::section("Migrations");
258
259 for (i, migration) in migrations.iter().enumerate() {
260 let name = migration.file_name().unwrap().to_string_lossy();
261 let applied = is_migration_applied(migration)?;
262
263 let status = if applied {
264 output::style_success("✓ Applied")
265 } else {
266 output::style_pending("○ Pending")
267 };
268
269 output::numbered_item(i + 1, &format!("{} - {}", name, status));
270 }
271
272 output::newline();
273
274 let applied_count = migrations
275 .iter()
276 .filter(|m| is_migration_applied(m).unwrap_or(false))
277 .count();
278 let pending_count = migrations.len() - applied_count;
279
280 output::kv("Total", &migrations.len().to_string());
281 output::kv("Applied", &applied_count.to_string());
282 output::kv("Pending", &pending_count.to_string());
283
284 Ok(())
285}
286
287async fn run_resolve(args: crate::cli::MigrateResolveArgs) -> CliResult<()> {
289 output::header("Migrate Resolve");
290
291 if args.rolled_back {
292 output::step(1, 2, "Marking migration as rolled back...");
293 output::step(2, 2, "Updating migration history...");
296
297 output::newline();
298 success(&format!(
299 "Migration '{}' marked as rolled back",
300 args.migration
301 ));
302 } else if args.applied {
303 output::step(1, 2, "Marking migration as applied...");
304 output::step(2, 2, "Updating migration history...");
307
308 output::newline();
309 success(&format!("Migration '{}' marked as applied", args.migration));
310 } else {
311 return Err(CliError::Command(
312 "Must specify --applied or --rolled-back".to_string(),
313 ));
314 }
315
316 Ok(())
317}
318
319async fn run_diff(args: crate::cli::MigrateDiffArgs) -> CliResult<()> {
321 output::header("Migrate Diff");
322
323 let _cwd = std::env::current_dir()?;
324
325 output::step(1, 3, "Parsing schema...");
327 let loaded = crate::schema_loader::load_schema(args.schema.as_deref())?;
328 let schema = loaded.schema;
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 check_pending_migrations(migrations_dir: &Path) -> CliResult<Vec<PathBuf>> {
443 let mut pending = Vec::new();
444
445 if !migrations_dir.exists() {
446 return Ok(pending);
447 }
448
449 for entry in std::fs::read_dir(migrations_dir)? {
450 let entry = entry?;
451 let path = entry.path();
452 if path.is_dir() && !is_migration_applied(&path)? {
453 pending.push(path);
454 }
455 }
456
457 pending.sort();
458 Ok(pending)
459}
460
461fn is_migration_applied(migration_path: &Path) -> CliResult<bool> {
462 let marker = migration_path.join(".applied");
465 Ok(marker.exists())
466}
467
468fn create_migration(
469 migrations_dir: &Path,
470 name: &str,
471 schema: &prax_schema::ast::Schema,
472) -> CliResult<PathBuf> {
473 let timestamp = chrono::Utc::now().format("%Y%m%d%H%M%S");
475 let migration_name = format!("{}_{}", timestamp, name);
476 let migration_path = migrations_dir.join(&migration_name);
477
478 std::fs::create_dir_all(&migration_path)?;
479
480 let sql = generate_schema_diff(schema)?;
482
483 let sql_path = migration_path.join("migration.sql");
485 std::fs::write(&sql_path, &sql)?;
486
487 Ok(migration_path)
488}
489
490fn generate_schema_diff(schema: &prax_schema::ast::Schema) -> CliResult<String> {
491 use prax_schema::ast::{FieldType, ScalarType};
492
493 let mut sql = String::new();
494
495 sql.push_str("-- Migration generated by Prax\n\n");
496
497 if !schema.enums.is_empty() {
499 sql.push_str("-- Enum types\n");
500 for enum_def in schema.enums.values() {
501 let enum_name = enum_def
502 .attributes
503 .iter()
504 .find(|a| a.is("map"))
505 .and_then(|a: &prax_schema::ast::Attribute| a.first_arg())
506 .and_then(|v: &prax_schema::ast::AttributeValue| v.as_string())
507 .map(|s| s.to_string())
508 .unwrap_or_else(|| to_snake_case(enum_def.name()));
509
510 sql.push_str(&format!(
511 "DO $$ BEGIN\n CREATE TYPE \"{}\" AS ENUM (",
512 enum_name
513 ));
514
515 let variants: Vec<String> = enum_def
516 .variants
517 .iter()
518 .map(|v| format!("'{}'", v.name()))
519 .collect();
520
521 sql.push_str(&variants.join(", "));
522 sql.push_str(");\nEXCEPTION\n WHEN duplicate_object THEN null;\nEND $$;\n\n");
523 }
524 sql.push('\n');
525 }
526
527 sql.push_str("-- Tables\n");
529 for model in schema.models.values() {
530 let table_name = model.table_name();
531
532 sql.push_str(&format!(
533 "CREATE TABLE IF NOT EXISTS \"{}\" (\n",
534 table_name
535 ));
536
537 let mut columns = Vec::new();
538 let mut primary_keys = Vec::new();
539
540 for field in model.fields.values() {
541 if field.is_relation() {
542 continue;
543 }
544
545 let column_name = field
546 .get_attribute("map")
547 .and_then(|a| a.first_arg())
548 .and_then(|v| v.as_string())
549 .map(|s| s.to_string())
550 .unwrap_or_else(|| to_snake_case(field.name()));
551
552 let sql_type = field_type_to_sql(&field.field_type);
553 let mut column_def = format!(" \"{}\" {}", column_name, sql_type);
554
555 if field.is_id() {
557 primary_keys.push(column_name.clone());
558 }
559
560 if field.has_attribute("auto") || field.has_attribute("autoincrement") {
561 column_def = format!(" \"{}\" SERIAL", column_name);
563 }
564
565 if field.has_attribute("unique") {
566 column_def.push_str(" UNIQUE");
567 }
568
569 if !field.is_optional() && !field.is_id() {
570 column_def.push_str(" NOT NULL");
571 }
572
573 if let Some(default_attr) = field.get_attribute("default")
575 && let Some(value) = default_attr.first_arg()
576 {
577 let value_str = format_attribute_value(value);
578 column_def.push_str(&format!(
579 " DEFAULT {}",
580 sql_default_value(&value_str, &field.field_type)
581 ));
582 }
583
584 columns.push(column_def);
585 }
586
587 sql.push_str(&columns.join(",\n"));
588
589 if !primary_keys.is_empty() {
590 sql.push_str(",\n");
591 sql.push_str(&format!(
592 " PRIMARY KEY (\"{}\")",
593 primary_keys.join("\", \"")
594 ));
595 }
596
597 sql.push_str("\n);\n\n");
598 }
599
600 return Ok(sql);
601
602 fn field_type_to_sql(field_type: &FieldType) -> String {
603 match field_type {
604 FieldType::Scalar(scalar) => match scalar {
605 ScalarType::Int => "INTEGER".to_string(),
606 ScalarType::BigInt => "BIGINT".to_string(),
607 ScalarType::Float => "DOUBLE PRECISION".to_string(),
608 ScalarType::String => "TEXT".to_string(),
609 ScalarType::Boolean => "BOOLEAN".to_string(),
610 ScalarType::DateTime => "TIMESTAMP WITH TIME ZONE".to_string(),
611 ScalarType::Date => "DATE".to_string(),
612 ScalarType::Time => "TIME".to_string(),
613 ScalarType::Json => "JSONB".to_string(),
614 ScalarType::Bytes => "BYTEA".to_string(),
615 ScalarType::Decimal => "DECIMAL".to_string(),
616 ScalarType::Uuid => "UUID".to_string(),
617 ScalarType::Cuid | ScalarType::Cuid2 | ScalarType::NanoId | ScalarType::Ulid => {
618 "TEXT".to_string()
619 }
620 ScalarType::Vector(dim) => match dim {
621 Some(d) => format!("vector({})", d),
622 None => "vector".to_string(),
623 },
624 ScalarType::HalfVector(dim) => match dim {
625 Some(d) => format!("halfvec({})", d),
626 None => "halfvec".to_string(),
627 },
628 ScalarType::SparseVector(dim) => match dim {
629 Some(d) => format!("sparsevec({})", d),
630 None => "sparsevec".to_string(),
631 },
632 ScalarType::Bit(dim) => match dim {
633 Some(d) => format!("bit({})", d),
634 None => "bit".to_string(),
635 },
636 },
637 FieldType::Enum(name) => format!("\"{}\"", to_snake_case(name)),
638 _ => "TEXT".to_string(),
639 }
640 }
641}
642
643async fn apply_migration(migration_path: &Path, _config: &Config) -> CliResult<()> {
644 let sql_path = migration_path.join("migration.sql");
645
646 if !sql_path.exists() {
647 return Err(CliError::Migration(format!(
648 "Migration file not found: {}",
649 sql_path.display()
650 )));
651 }
652
653 let _sql = std::fs::read_to_string(&sql_path)?;
654
655 let marker = migration_path.join(".applied");
660 std::fs::write(&marker, chrono::Utc::now().to_rfc3339())?;
661
662 Ok(())
663}
664
665fn sql_default_value(value: &str, field_type: &prax_schema::ast::FieldType) -> String {
666 if matches!(field_type, prax_schema::ast::FieldType::Enum(_)) {
668 return format!("'{}'", value);
669 }
670
671 match value.to_lowercase().as_str() {
672 "now()" => "CURRENT_TIMESTAMP".to_string(),
673 "uuid()" => "gen_random_uuid()".to_string(),
674 "cuid()" | "cuid2()" | "nanoid()" | "ulid()" => {
675 "''".to_string()
677 }
678 "true" => "TRUE".to_string(),
679 "false" => "FALSE".to_string(),
680 _ => value.to_string(),
681 }
682}
683
684fn to_snake_case(name: &str) -> String {
685 let mut result = String::new();
686 for (i, c) in name.chars().enumerate() {
687 if c.is_uppercase() {
688 if i > 0 {
689 result.push('_');
690 }
691 result.push(c.to_lowercase().next().unwrap());
692 } else {
693 result.push(c);
694 }
695 }
696 result
697}
698
699fn format_attribute_value(value: &prax_schema::ast::AttributeValue) -> String {
700 use prax_schema::ast::AttributeValue;
701
702 match value {
703 AttributeValue::String(s) => format!("\"{}\"", s),
704 AttributeValue::Int(i) => i.to_string(),
705 AttributeValue::Float(f) => f.to_string(),
706 AttributeValue::Boolean(b) => b.to_string(),
707 AttributeValue::Ident(id) => id.to_string(),
708 AttributeValue::Function(name, args) => {
709 if args.is_empty() {
710 format!("{}()", name)
711 } else {
712 let arg_strs: Vec<String> = args.iter().map(format_attribute_value).collect();
713 format!("{}({})", name, arg_strs.join(", "))
714 }
715 }
716 AttributeValue::Array(items) => {
717 let item_strs: Vec<String> = items.iter().map(format_attribute_value).collect();
718 format!("[{}]", item_strs.join(", "))
719 }
720 AttributeValue::FieldRef(field) => field.to_string(),
721 AttributeValue::FieldRefList(fields) => {
722 format!(
723 "[{}]",
724 fields
725 .iter()
726 .map(|f| f.to_string())
727 .collect::<Vec<_>>()
728 .join(", ")
729 )
730 }
731 }
732}