Branches: Merge (#1253)

* add merge command

* print "Done!"

* clippy

* Update src/migrations/merge.rs

Co-authored-by: Aljaž Mur Eržen <aljazerzen@users.noreply.github.com>

* update error messages to be more clear

* change up error message

* remove redundant id updates

---------

Co-authored-by: Aljaž Mur Eržen <aljazerzen@users.noreply.github.com>

src/branch/main.rs

@@ -1,6 +1,6 @@
 use crate::branch::context::Context;
 use crate::branch::option::{BranchCommand, Command};
-use crate::branch::{create, drop, list, rebase, rename, switch, wipe};
+use crate::branch::{create, drop, list, merge, rebase, rename, switch, wipe};
 use crate::connect::{Connection, Connector};
 use crate::options::Options;
 
@@ -26,6 +26,7 @@ pub async fn branch_main(options: &Options, cmd: &BranchCommand) -> anyhow::Resu
                 Command::List(list) => list::main(list, &context, &mut connection).await,
                 Command::Rename(rename) => rename::main(rename, &context, &mut connection, &options).await,
                 Command::Rebase(rebase) => rebase::main(rebase, &context, &mut connection, &options).await,
+                Command::Merge(merge) => merge::main(merge, &context, &mut connection, &options).await,
                 unhandled => anyhow::bail!("unimplemented branch command '{:?}'", unhandled)
             }
         }

src/branch/merge.rs

@@ -0,0 +1,36 @@
+use crate::branch::connections::connect_if_branch_exists;
+use crate::branch::context::Context;
+use crate::branch::option::{Merge};
+use crate::connect::Connection;
+use crate::migrations;
+use crate::migrations::merge::{apply_merge_migration_files, get_merge_migrations, write_merge_migrations};
+use crate::options::Options;
+
+pub async fn main(options: &Merge, context: &Context, source_connection: &mut Connection, cli_opts: &Options) -> anyhow::Result<()> {
+    if options.target_branch == context.branch {
+        anyhow::bail!("Cannot merge the current branch into its self");
+    }
+
+    let mut target_connection = match connect_if_branch_exists(
+        cli_opts.create_connector().await?.database(&options.target_branch)?
+    ).await? {
+        Some(connection) => connection,
+        None => anyhow::bail!("The branch '{}' doesn't exist", options.target_branch)
+    };
+
+    let migration_context = migrations::Context::for_project(&context.project_config)?;
+    let mut merge_migrations = get_merge_migrations(source_connection, &mut target_connection).await?;
+
+    eprintln!("Merging {} migration(s) into '{}'...", merge_migrations.target_migrations.len(), source_connection.database());
+
+    write_merge_migrations(&migration_context, &mut merge_migrations).await?;
+
+    if !options.no_apply {
+        eprintln!("Applying migrations...");
+        apply_merge_migration_files(&merge_migrations, &migration_context, source_connection).await?;
+    }
+
+    eprintln!("Done!");
+
+    Ok(())
+}

src/branch/mod.rs

@@ -9,5 +9,6 @@ pub mod option;
 mod rename;
 mod switch;
 mod wipe;
+mod merge;
 
 pub use main::branch_main;

src/branch/option.rs

@@ -12,7 +12,8 @@ pub enum Command {
     Switch(Switch),
     Rename(Rename),
     List(List),
-    Rebase(Rebase)
+    Rebase(Rebase),
+    Merge(Merge)
 }
 
 /// Creates a new branch and switches to it.
@@ -115,3 +116,13 @@ pub struct Rebase {
     #[arg(long)]
     pub no_apply: bool,
 }
+
+#[derive(clap::Args, Clone, Debug)]
+pub struct Merge {
+    /// The branch to merge into this one
+    pub target_branch: String,
+
+    /// Skip applying migrations generated from the merge
+    #[arg(long)]
+    pub no_apply: bool,
+}

src/migrations/merge.rs

@@ -0,0 +1,151 @@
+use indexmap::IndexMap;
+use fs_err as fs;
+
+use crate::connect::Connection;
+use crate::migrations::{Context, migrate, migration};
+use crate::migrations::create::{MigrationKey, MigrationToText, write_migration};
+use crate::migrations::db_migration::{DBMigration, read_all};
+use crate::migrations::migration::MigrationFile;
+use crate::migrations::rebase::{fix_migration_ids};
+use crate::print;
+
+pub struct MergeMigrations {
+    pub base_migrations: IndexMap<String, MergeMigration>,
+    pub target_migrations: IndexMap<String, MergeMigration>
+}
+
+impl MergeMigrations {
+    fn flatten(&self) -> IndexMap<&String, &MergeMigration> {
+        let mut result = IndexMap::new();
+
+        for migration in self.base_migrations.iter() {
+            result.insert(migration.0, migration.1);
+        }
+
+        for migration in self.target_migrations.iter() {
+            result.insert(migration.0, migration.1);
+        }
+
+        result
+    }
+}
+
+pub struct MergeMigration {
+    key: MigrationKey,
+    migration: DBMigration
+}
+
+impl MigrationToText for MergeMigration {
+    type StatementsIter<'a> = std::iter::Once<&'a String> where Self: 'a;
+
+    fn key(&self) -> &MigrationKey {
+        &self.key
+    }
+
+    fn parent(&self) -> anyhow::Result<&str> {
+        Ok(self.migration.parent_names.first().map(String::as_str).unwrap_or("initial"))
+    }
+
+    fn id(&self) -> anyhow::Result<&str> {
+        Ok(self.migration.name.as_str())
+    }
+
+    fn statements<'a>(&'a self) -> Self::StatementsIter<'a> {
+        std::iter::once(&self.migration.script)
+    }
+}
+
+pub async fn get_merge_migrations(base: &mut Connection, target: &mut Connection) -> anyhow::Result<MergeMigrations> {
+    let base_migrations = read_all(base, true, false).await?;
+    let mut target_migrations = read_all(target, true, false).await?;
+
+    // check if the base branch is up-to-date with target:
+    // we do this by verifying that all target migrations
+    // exist within the base migrations
+    if base_migrations.len() >= target_migrations.len() &&
+        target_migrations.iter().enumerate()
+            .all(|(i, (k, _))|
+                base_migrations.get_index(i)
+                    .map(|v| v.0 == k).unwrap_or(false)
+            )
+    {
+        anyhow::bail!("Already up to date.")
+    }
+
+    let mut diverging_migrations = Vec::new();
+
+    for (index, (base_migration_id, _)) in base_migrations.iter().enumerate() {
+        if let Some((target_migration_id, _)) = target_migrations.get_index(index) {
+            if target_migration_id != base_migration_id {
+                diverging_migrations.push((index, base_migration_id, target_migration_id))
+            }
+        }
+    }
+
+    if diverging_migrations.len() > 0 {
+        eprintln!("\nThe migration history of {} diverges from {}:", target.database(), base.database());
+
+        for (index, expecting, actual) in &diverging_migrations {
+            eprintln!("{}. Expecting {} but has {}", index, &expecting[..7], &actual[..7])
+        }
+
+        eprintln!();
+
+        anyhow::bail!("Cannot complete fast-forward merge, the histories of {0} and {1} are incompatible. Try rebasing {1} onto {0}", base.database(), target.database())
+    }
+
+    let mut target_merge_migrations: IndexMap<String, MergeMigration> = IndexMap::new();
+    let mut base_merge_migrations: IndexMap<String, MergeMigration> = IndexMap::new();
+
+    for (index, (id, migration)) in target_migrations.split_off(base_migrations.len()).into_iter().enumerate() {
+        target_merge_migrations.insert(id, MergeMigration {
+            migration,
+            key: MigrationKey::Index((base_migrations.len() + index + 1) as u64),
+        });
+    }
+
+    for(index, (id, migration)) in target_migrations.into_iter().enumerate() {
+        base_merge_migrations.insert(id, MergeMigration {
+            migration,
+            key: MigrationKey::Index((index + 1) as u64),
+        });
+    }
+
+    Ok(MergeMigrations {
+        target_migrations: target_merge_migrations,
+        base_migrations: base_merge_migrations
+    })
+}
+
+pub async fn write_merge_migrations(context: &Context, migrations: &mut MergeMigrations) -> anyhow::Result<()> {
+    let temp_dir = tempfile::tempdir()?;
+    let temp_ctx = Context {
+        schema_dir: temp_dir.path().to_path_buf(),
+        edgedb_version: None,
+        quiet: false,
+    };
+
+    for (_, migration) in migrations.flatten() {
+        write_migration(&temp_ctx, migration, false).await?;
+    }
+
+    for from in migration::read_names(&temp_ctx).await? {
+        let to = context
+            .schema_dir
+            .join("migrations")
+            .join(from.file_name().unwrap());
+
+        fs::copy(from, to)?;
+    }
+
+    Ok(())
+}
+
+pub async fn apply_merge_migration_files(merge_migrations: &MergeMigrations, context: &Context, connection: &mut Connection) -> anyhow::Result<()> {
+    // apply the new migrations
+    let migrations: IndexMap<String, MigrationFile> = migration::read_all(context, true).await?.into_iter()
+        .filter(|(id, _)| merge_migrations.target_migrations.contains_key(id))
+        .collect();
+
+    migrate::apply_migrations(connection, &migrations, context, true).await
+}

src/migrations/mod.rs

@@ -19,6 +19,7 @@ pub mod dev_mode;
 pub mod options;
 pub mod upgrade_check;
 pub mod rebase;
+pub mod merge;
 
 const NULL_MIGRATION: &str = "initial";
 

src/migrations/rebase.rs

@@ -10,10 +10,6 @@ use crate::connect::Connection;
 use crate::migrations::{Context, create, migrate, migration};
 use crate::migrations::create::{MigrationKey, MigrationToText};
 use crate::migrations::db_migration::{DBMigration, read_all};
-
-
-
-
 use crate::migrations::migration::MigrationFile;
 use crate::print;
 
@@ -189,6 +185,22 @@ pub async fn get_diverging_migrations(source: &mut Connection, target: &mut Conn
 }
 
 async fn rebase_migration_ids(context: &Context, rebase_migrations: &mut RebaseMigrations) -> anyhow::Result<()> {
+    fn update_id(old: &String, new: &String, col: &mut IndexMap<String, DBMigration>) {
+        if let Some(value) = col.remove(old) {
+            col.insert(new.clone(), value);
+        }
+    }
+
+    fix_migration_ids(context, |old, new| {
+        update_id(old, new, &mut rebase_migrations.base_migrations);
+        update_id(old, new, &mut rebase_migrations.target_migrations);
+        update_id(old, new, &mut rebase_migrations.source_migrations);
+    }).await
+}
+
+pub async fn fix_migration_ids<T>(context: &Context, mut on_update: T) -> anyhow::Result<()>
+    where T : FnMut(&String, &String)
+{
     let migrations = migration::read_all(context, false).await?;
     let mut changed_ids: HashMap<String, String> = HashMap::new();
 
@@ -213,17 +225,8 @@ async fn rebase_migration_ids(context: &Context, rebase_migrations: &mut RebaseM
         fs::write(&migration_file.path, migration_text)?;
     }
 
-    fn update_id(old: &String, new: &String, col: &mut IndexMap<String, DBMigration>) {
-        if let Some(value) = col.remove(old) {
-            col.insert(new.clone(), value);
-        }
-    }
-
-    // update rebase_migrations to match the updated IDs
     for (old_id, new_id) in changed_ids {
-        update_id(&old_id, &new_id, &mut rebase_migrations.target_migrations);
-        update_id(&old_id, &new_id, &mut rebase_migrations.source_migrations);
-        update_id(&old_id, &new_id, &mut rebase_migrations.base_migrations);
+        on_update(&old_id, &new_id);
     }
 
     Ok(())