Repository: edgedb/edgedb-cli (rust)

Issue:
Installer version errors or unexpected behavior
When running the edgedb cli installer with the `--version` option, an error results. I expected information about the version of the cli that would be installed.

(I encountered this while attempting to determine and select which version of the cli would be installed, as I had begun encountering some errors after upgrading from `2.2.6` to `2.3.0`, and wanted to determine if those errors were version related or otherwise.)

To reproduce:
```
curl --proto '=https' --tlsv1.2 -sSf https://sh.edgedb.com > edgedb-cli-installer
chmod u+x edgedb-cli-installer
./edgedb-cli-installer --help
```
  <details><summary>Output includes: <code>--version    Print version information`</code></summary>
  <p>

```
edgedb-init
The installer for edgedb command-line tools

USAGE:
    edgedb-init [FLAGS] [OPTIONS]

FLAGS:
    -h, --help              Print help information
        --nightly           Install nightly version of command-line tools
    -y                      Disable confirmation prompt
        --no-modify-path    Do not configure the PATH environment variable
    -q, --quiet             Disable progress output
    -v, --verbose           Enable verbose output
    -V, --version           Print version information
```
</p>
</details>


`./edgedb-cli-installer --version`

<details><summary>Output includes: <code>error: Found argument '--version' which wasn't expected, or isn't valid in this context</code></summary><p>

```
info: downloading installer
error: Found argument '--version' which wasn't expected, or isn't valid in this context

	Did you mean '--verbose'?

	If you tried to supply `--version` as a value rather than a flag, use `-- --version`

USAGE:
    edgedb _self_install --verbose

For more information try --help
```
</p></details>

[^1]

[^1]:OS Version: macOS 13.0.1 (22A400)

Pull request:
edgedb-init: Remove -V,--version param from help
This closes #945

Patch:
diff --git a/Cargo.lock b/Cargo.lock
index 0bc527c..72fd9d5 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1102,7 +1102,7 @@ dependencies = [
 [[package]]
 name = "edgeql-parser"
 version = "0.1.0"
-source = "git+https://github.com/edgedb/edgedb#d92aedfefc6e83c9d6ffefbfab79f15e9e9fe124"
+source = "git+https://github.com/edgedb/edgedb#77c290452a67d5f6c09ef385600af41eade83f2d"
 dependencies = [
  "base32",
  "combine",
diff --git a/edgedb-init.sh b/edgedb-init.sh
index 7ee6912..80baf4d 100755
--- a/edgedb-init.sh
+++ b/edgedb-init.sh
@@ -27,7 +27,6 @@ FLAGS:
         --no-modify-path    Do not configure the PATH environment variable
     -q, --quiet             Disable progress output
     -v, --verbose           Enable verbose output
-    -V, --version           Print version information
 EOF
 }
 
diff --git a/src/async_util.rs b/src/async_util.rs
index 61dda42..3e55201 100644
--- a/src/async_util.rs
+++ b/src/async_util.rs
@@ -10,3 +10,21 @@ pub async fn timeout<F, T, E>(dur: Duration, f: F) -> anyhow::Result<T>
     .map(|r| r.map_err(Into::into))
     .unwrap_or_else(|_| Err(io::Error::from(io::ErrorKind::TimedOut).into()))
 }
+
+#[macro_export]
+macro_rules! async_try {
+    ($block:expr, finally $finally:expr) => {
+        {
+            let result = $block.await;
+            match ($finally.await, result) {
+                (Ok(()), Ok(res)) => Ok(res),
+                (Err(e), Ok(_)) => Err(e.into()),
+                (Ok(()), Err(e)) => Err(e),
+                (Err(fin_e), Err(e)) => {
+                    log::info!("Cannot finalize operation: {:#}", fin_e);
+                    Err(e)
+                }
+            }
+        }
+    }
+}
diff --git a/src/highlight.rs b/src/highlight.rs
index 2d6b097..3dd67da 100644
--- a/src/highlight.rs
+++ b/src/highlight.rs
@@ -134,6 +134,7 @@ fn token_style(kind: Kind, value: &str) -> Option<Style> {
         T::Colon => Some(S::Operator),
         T::Add => Some(S::Operator),
         T::Sub => Some(S::Operator),
+        T::DoubleSplat => Some(S::Operator),
         T::Mul => Some(S::Operator),
         T::Div => Some(S::Operator),
         T::Modulo => Some(S::Operator),
diff --git a/src/interactive.rs b/src/interactive.rs
index e35d0e4..f00ac26 100644
--- a/src/interactive.rs
+++ b/src/interactive.rs
@@ -252,6 +252,13 @@ async fn execute_backslash(mut state: &mut repl::State, text: &str)
     Ok(())
 }
 
+async fn write_out(data: &str) -> anyhow::Result<()> {
+    let mut out = stdout();
+    out.write_all(data.as_bytes()).await?;
+    out.flush().await?;
+    Ok(())
+}
+
 async fn execute_query(options: &Options, state: &mut repl::State,
     statement: &str)
     -> anyhow::Result<()>
@@ -372,8 +379,7 @@ async fn execute_query(options: &Options, state: &mut repl::State,
                 };
                 // trying to make writes atomic if possible
                 text += "\n";
-                stdout().write_all(text.as_bytes()).await?;
-                stdout().flush().await?;
+                write_out(&text).await?;
                 index += 1;
             }
         }
@@ -429,8 +435,7 @@ async fn execute_query(options: &Options, state: &mut repl::State,
                 // trying to make writes atomic if possible
                 let mut data = print::json_to_string(jitems, &cfg)?;
                 data += "\n";
-                stdout().write_all(data.as_bytes()).await?;
-                stdout().flush().await?;
+                write_out(&data).await?;
             }
         }
         JsonPretty | JsonLines => {
@@ -465,15 +470,13 @@ async fn execute_query(options: &Options, state: &mut repl::State,
                 if state.output_format == JsonLines {
                     // trying to make writes atomic if possible
                     text += "\n";
-                    stdout().write_all(text.as_bytes()).await?;
-                    stdout().flush().await?;
+                    write_out(&text).await?;
                 } else {
                     // trying to make writes atomic if possible
                     let mut data;
                     data = print::json_item_to_string(&value, &cfg)?;
                     data += "\n";
-                    stdout().write_all(data.as_bytes()).await?;
-                    stdout().flush().await?;
+                    write_out(&data).await?;
                     index += 1;
                 }
             }
diff --git a/src/log_levels.rs b/src/log_levels.rs
index f5c1aab..4cbff94 100644
--- a/src/log_levels.rs
+++ b/src/log_levels.rs
@@ -26,6 +26,10 @@ pub fn init(builder: &mut env_logger::Builder, opt: &Options) {
         Some(Command::Common(Common::Restore(r))) if r.verbose => {
             builder.filter_module("edgedb::restore", log::LevelFilter::Info);
         }
+        Some(Command::Watch(w)) if w.verbose => {
+            builder.filter_module("edgedb::migrations::dev_mode::ddl",
+                                  log::LevelFilter::Info);
+        }
         Some(Command::Common(Common::Migration(c))) => match &c.subcommand {
             MigrationCmd::Create(c) if c.debug_print_queries => {
                 builder.filter_module("edgedb::migrations::query",
diff --git a/src/migrations/create.rs b/src/migrations/create.rs
index 6d321c0..ce46fff 100644
--- a/src/migrations/create.rs
+++ b/src/migrations/create.rs
@@ -5,7 +5,7 @@ use std::path::{Path, PathBuf};
 use colorful::Colorful;
 use crate::connect::Connection;
 use edgedb_derive::Queryable;
-use edgedb_errors::{Error, QueryError};
+use edgedb_errors::{Error, QueryError, InvalidSyntaxError};
 use edgedb_errors::{ErrorKind, ClientConnectionEosError};
 use edgedb_protocol::queryable::Queryable;
 use edgeql_parser::expr;
@@ -21,6 +21,7 @@ use tokio::fs;
 use tokio::io::{self, AsyncWriteExt};
 use tokio::task::spawn_blocking as unblock;
 
+use crate::async_try;
 use crate::bug;
 use crate::commands::parser::CreateMigration;
 use crate::commands::{Options, ExitCode};
@@ -62,6 +63,12 @@ enum Choice {
 pub struct RequiredUserInput {
     placeholder: String,
     prompt: String,
+    #[allow(dead_code)]
+    old_type: Option<String>,
+    new_type: Option<String>,
+    #[serde(rename="type")]
+    type_name: Option<String>,
+    pointer_name: Option<String>,
 }
 
 #[derive(Deserialize, Debug)]
@@ -104,6 +111,12 @@ struct Refused;
 #[error("split migration")]
 struct SplitMigration;
 
+#[derive(Debug, thiserror::Error)]
+#[error("EdgeDB could not resolve migration automatically. \
+         Please run `edgedb migration create` in interactive mode.")]
+struct CantResolve;
+
+
 pub async fn execute(cli: &mut Connection, text: impl AsRef<str>)
     -> Result<(), Error>
 {
@@ -111,6 +124,21 @@ pub async fn execute(cli: &mut Connection, text: impl AsRef<str>)
         return Err(ClientConnectionEosError::with_message(
                 "connection closed by server"));
     }
+    log_execute(cli, text).await
+}
+
+pub async fn execute_if_connected(cli: &mut Connection, text: impl AsRef<str>)
+    -> Result<(), Error>
+{
+    if !cli.is_consistent() {
+        return Ok(())
+    }
+    log_execute(cli, text).await
+}
+
+async fn log_execute(cli: &mut Connection, text: impl AsRef<str>)
+    -> Result<(), Error>
+{
     let text = text.as_ref();
     log::debug!(target: "edgedb::migrations::query", "Executing `{}`", text);
     cli.execute(text, &()).await?;
@@ -239,19 +267,111 @@ async fn first_migration(cli: &mut Connection, ctx: &Context,
     }
 }
 
-async fn run_non_interactive(ctx: &Context, cli: &mut Connection, index: u64,
-    options: &CreateMigration)
-    -> anyhow::Result<()>
+pub async fn unsafe_populate(_ctx: &Context, cli: &mut Connection)
+    -> anyhow::Result<CurrentMigration>
 {
-    let descr = loop {
+    loop {
         let data = query_row::<CurrentMigration>(cli,
             "DESCRIBE CURRENT MIGRATION AS JSON"
         ).await?;
         if data.complete {
-            break data;
+            return Ok(data);
+        }
+        if let Some(proposal) = &data.proposed {
+            let mut placeholders = BTreeMap::new();
+            if !proposal.required_user_input.is_empty() {
+                for input in &proposal.required_user_input {
+                    let name = &input.placeholder[..];
+                    let kind_end = name.find("_expr").unwrap_or(name.len());
+                    let expr = match &name[..kind_end] {
+                        "fill" if input.type_name.is_some() => {
+                            format!("<{}>{{}}",
+                                    input.type_name.as_ref().unwrap())
+                        }
+                        "cast"
+                            if input.pointer_name.is_some() &&
+                               input.new_type.is_some()
+                        => {
+                            format!("<{}>.{}",
+                                    input.new_type.as_ref().unwrap(),
+                                    input.pointer_name.as_ref().unwrap())
+                        }
+                        "conv" if input.pointer_name.is_some() => {
+                            format!("(SELECT .{} LIMIT 1)",
+                                    input.pointer_name.as_ref().unwrap())
+                        }
+                        _ => {
+                            log::debug!("Cannot fill placeholder {} \
+                                        into {:?}, input info: {:?}",
+                                        input.placeholder,
+                                        proposal.statements,
+                                        input);
+                            return Err(CantResolve)?;
+                        }
+                    };
+                    placeholders.insert(input.placeholder.clone(), expr);
+                }
+            }
+            if !apply_proposal(cli, &proposal, &placeholders).await? {
+                execute(cli, "ALTER CURRENT MIGRATION REJECT PROPOSED").await?;
+            }
+        } else {
+            log::debug!("No proposal generated");
+            return Err(CantResolve)?;
+        }
+    }
+}
+
+async fn apply_proposal(cli: &mut Connection, proposal: &Proposal,
+                        placeholders: &BTreeMap<String, String>)
+    -> anyhow::Result<bool>
+{
+    execute(cli, "DECLARE SAVEPOINT proposal").await?;
+    let mut rollback = false;
+    async_try!{
+        async {
+            for statement in &proposal.statements {
+                let statement = substitute_placeholders(
+                    &statement.text, &placeholders)?;
+                if let Err(e) = execute(cli, &statement).await {
+                    if e.is::<InvalidSyntaxError>() {
+                        log::error!("Error executing: {}", statement);
+                        return Err(e)?;
+                    } else if e.is::<QueryError>() {
+                        rollback = true;
+                        log::info!("Statement {:?} failed: {:#}",
+                                   statement, e);
+                        return Ok(false);
+                    } else {
+                        return Err(e)?;
+                    }
+                }
+            }
+            Ok(true)
+        },
+        finally async {
+            if rollback {
+                execute_if_connected(cli,
+                    "ROLLBACK TO SAVEPOINT proposal",
+                ).await?;
+            }
+            execute_if_connected(cli, "RELEASE SAVEPOINT proposal").await
+        }
+    }
+}
+
+async fn non_interactive_populate(_ctx: &Context, cli: &mut Connection)
+    -> anyhow::Result<CurrentMigration>
+{
+    loop {
+        let data = query_row::<CurrentMigration>(cli,
+            "DESCRIBE CURRENT MIGRATION AS JSON"
+        ).await?;
+        if data.complete {
+            return Ok(data);
         }
         if let Some(proposal) = data.proposed {
-            if proposal.confidence >= SAFE_CONFIDENCE || options.allow_unsafe {
+            if proposal.confidence >= SAFE_CONFIDENCE {
                 if !proposal.required_user_input.is_empty() {
                     for input in proposal.required_user_input {
                         eprintln!("Input required: {}", input.prompt);
@@ -280,6 +400,17 @@ async fn run_non_interactive(ctx: &Context, cli: &mut Connection, index: u64,
                 interactive mode to confirm changes, \
                 or use `--allow-unsafe`");
         }
+    }
+}
+
+async fn run_non_interactive(ctx: &Context, cli: &mut Connection, index: u64,
+    options: &CreateMigration)
+    -> anyhow::Result<()>
+{
+    let descr = if options.allow_unsafe {
+        unsafe_populate(ctx, cli).await?
+    } else {
+        non_interactive_populate(ctx, cli).await?
     };
     if descr.confirmed.is_empty() && !options.allow_empty {
         print::warn("No schema changes detected.");
@@ -600,38 +731,36 @@ pub async fn normal_migration(cli: &mut Connection, ctx: &Context,
     -> anyhow::Result<()>
 {
     execute_start_migration(&ctx, cli).await?;
-    let exec = async {
-        let descr = query_row::<CurrentMigration>(cli,
-            "DESCRIBE CURRENT MIGRATION AS JSON",
-        ).await?;
-        let db_migration = if descr.parent == "initial" {
-            None
-        } else {
-            Some(&descr.parent)
-        };
-        if db_migration != migrations.keys().last() {
-            anyhow::bail!("Database must be updated to the last migration \
-                on the filesystem for `migration create`. Run:\n  \
-                edgedb migrate");
-        }
+    async_try! {
+        async {
+            let descr = query_row::<CurrentMigration>(cli,
+                "DESCRIBE CURRENT MIGRATION AS JSON",
+            ).await?;
+            let db_migration = if descr.parent == "initial" {
+                None
+            } else {
+                Some(&descr.parent)
+            };
+            if db_migration != migrations.keys().last() {
+                anyhow::bail!("Database must be updated to the last migration \
+                    on the filesystem for `migration create`. Run:\n  \
+                    edgedb migrate");
+            }
 
-        let num_migrations = migrations.len() as u64 +1;
-        if create.non_interactive {
-            run_non_interactive(&ctx, cli, num_migrations, &create).await
-        } else {
-            if create.allow_unsafe {
-                log::warn!("The `--allow-unsafe` flag is unused \
-                            in interactive mode");
+            let num_migrations = migrations.len() as u64 +1;
+            if create.non_interactive {
+                run_non_interactive(&ctx, cli, num_migrations, &create).await
+            } else {
+                if create.allow_unsafe {
+                    log::warn!("The `--allow-unsafe` flag is unused \
+                                in interactive mode");
+                }
+                run_interactive(&ctx, cli, num_migrations, &create).await
             }
-            run_interactive(&ctx, cli, num_migrations, &create).await
+        },
+        finally async {
+            execute_if_connected(cli, "ABORT MIGRATION").await
         }
-    }.await;
-    if cli.is_consistent() {
-        let abort = cli.execute("ABORT MIGRATION", &()).await
-            .map(|_| ()).map_err(|e| e.into());
-        return exec.and(abort)
-    } else {
-        return exec
     }
 }
 
diff --git a/src/migrations/dev_mode.rs b/src/migrations/dev_mode.rs
index c9e9c31..f931af6 100644
--- a/src/migrations/dev_mode.rs
+++ b/src/migrations/dev_mode.rs
@@ -4,12 +4,15 @@ use linked_hash_map::LinkedHashMap;
 use anyhow::Context as _;
 use indicatif::ProgressBar;
 use once_cell::sync::Lazy;
+use edgedb_errors::{QueryError};
 
+use crate::async_try;
 use crate::commands::Options;
 use crate::commands::parser::CreateMigration;
 use crate::migrations::context::Context;
 use crate::migrations::create::{CurrentMigration, normal_migration};
 use crate::migrations::create::{execute, query_row, execute_start_migration};
+use crate::migrations::create::{execute_if_connected, unsafe_populate};
 use crate::migrations::migrate::{apply_migrations, apply_migrations_inner};
 use crate::migrations::migration::{self, MigrationFile};
 use crate::migrations::timeout;
@@ -127,21 +130,69 @@ async fn get_db_migration(cli: &mut Connection)
 async fn migrate_to_schema(cli: &mut Connection, ctx: &Context)
     -> anyhow::Result<()>
 {
-    execute_start_migration(&ctx, cli).await?;
-    let res = async {
-        execute(cli, "POPULATE MIGRATION").await?;
-        let descr = query_row::<CurrentMigration>(cli,
-            "DESCRIBE CURRENT MIGRATION AS JSON"
-        ).await?;
-        if !descr.complete {
-            // TODO(tailhook) is `POPULATE MIGRATION` equivalent to `--yolo` or
-            // should we do something manually?
-            anyhow::bail!("Migration cannot be automatically populated");
+    use edgedb_protocol::server_message::TransactionState::NotInTransaction;
+
+    let transaction = matches!(cli.transaction_state(), NotInTransaction);
+    if transaction {
+        execute(cli, "START TRANSACTION").await?;
+    }
+    async_try!{
+        async {
+            _migrate_to_schema(cli, ctx).await
+        },
+        finally async {
+            if transaction {
+                execute_if_connected(cli, "COMMIT").await?;
+            }
+            anyhow::Ok(())
         }
-        Ok(descr)
-    }.await;
-    execute(cli, "ABORT MIGRATION").await?;
-    let descr = res?;
+    }
+}
+
+async fn _migrate_to_schema(cli: &mut Connection, ctx: &Context)
+    -> anyhow::Result<()>
+{
+    execute(cli, "DECLARE SAVEPOINT migrate_to_schema").await?;
+    let descr = async_try! {
+        async {
+            execute_start_migration(&ctx, cli).await?;
+            if let Err(e) = execute(cli, "POPULATE MIGRATION").await {
+                if e.is::<QueryError>() {
+                    return Ok(None)
+                } else {
+                    return Err(e)?;
+                }
+            }
+            let descr = query_row::<CurrentMigration>(cli,
+                "DESCRIBE CURRENT MIGRATION AS JSON"
+            ).await?;
+            if !descr.complete {
+                anyhow::bail!("Migration cannot be automatically populated");
+            }
+            Ok(Some(descr))
+        },
+        finally async {
+            execute_if_connected(cli,
+                "ROLLBACK TO SAVEPOINT migrate_to_schema",
+            ).await?;
+            execute_if_connected(cli,
+                "RELEASE SAVEPOINT migrate_to_schema",
+            ).await
+        }
+    }?;
+    let descr = if let Some(descr) = descr {
+        descr
+    } else {
+        execute_start_migration(&ctx, cli).await?;
+        async_try! {
+            async {
+                unsafe_populate(ctx, cli).await
+            },
+            finally async {
+                execute_if_connected(cli, "ABORT MIGRATION",).await
+            }
+        }?
+    };
     if !descr.confirmed.is_empty() {
         ddl::apply_statements(cli, &descr.confirmed).await?;
     }
@@ -156,20 +207,7 @@ async fn rebase_to_schema(cli: &mut Connection, ctx: &Context,
 
     let res = async {
         apply_migrations_inner(cli, migrations, true).await?;
-        execute_start_migration(&ctx, cli).await?;
-        execute(cli, "POPULATE MIGRATION").await?;
-        let descr = query_row::<CurrentMigration>(cli,
-            "DESCRIBE CURRENT MIGRATION AS JSON"
-        ).await?;
-        if !descr.complete {
-            // TODO(tailhook) is `POPULATE MIGRATION` equivalent to `--yolo` or
-            // should we do something manually?
-            anyhow::bail!("Migration cannot be automatically populated");
-        }
-        execute(cli, "ABORT MIGRATION").await?;
-        if !descr.confirmed.is_empty() {
-            ddl::apply_statements(cli, &descr.confirmed).await?;
-        }
+        migrate_to_schema(cli, ctx).await?;
         Ok(())
     }.await;
 
@@ -180,7 +218,7 @@ async fn rebase_to_schema(cli: &mut Connection, ctx: &Context,
             Ok(())
         }
         Err(e) => {
-            execute(cli, "ABORT MIGRATION REWRITE").await
+            execute_if_connected(cli, "ABORT MIGRATION REWRITE").await
                 .map_err(|e| {
                     log::warn!("Error aborting migration rewrite: {:#}", e);
                 }).ok();
diff --git a/src/watch/options.rs b/src/watch/options.rs
index cdab9cf..910cd4d 100644
--- a/src/watch/options.rs
+++ b/src/watch/options.rs
@@ -3,4 +3,7 @@ use edgedb_cli_derive::{EdbClap};
 
 #[derive(EdbClap, Debug, Clone)]
 pub struct WatchCommand {
+    /// Print DDLs applied to the schema
+    #[clap(short='v', long)]
+    pub verbose: bool,
 }


Test patch:
diff --git a/tests/func/migrations.rs b/tests/func/migrations.rs
index 8cedde1..0ee37c2 100644
--- a/tests/func/migrations.rs
+++ b/tests/func/migrations.rs
@@ -722,3 +722,23 @@ fn dev_mode() -> anyhow::Result<()> {
         .assert().success();
     Ok(())
 }
+
+#[test]
+fn unsafe_migrations() -> anyhow::Result<()> {
+    SERVER.admin_cmd()
+        .arg("database").arg("create").arg("db_unsafe")
+        .assert().success();
+    SERVER.admin_cmd()
+        .arg("--database=db_unsafe")
+        .arg("migrate").arg("--dev-mode")
+        .arg("--schema-dir=tests/migrations/db_unsafe/initial")
+        .env("NO_COLOR", "1")
+        .assert().success();
+    SERVER.admin_cmd()
+        .arg("--database=db_unsafe")
+        .arg("migrate").arg("--dev-mode")
+        .arg("--schema-dir=tests/migrations/db_unsafe/modified1")
+        .env("NO_COLOR", "1")
+        .assert().success();
+    Ok(())
+}
diff --git a/tests/migrations/db_unsafe/initial/default.esdl b/tests/migrations/db_unsafe/initial/default.esdl
new file mode 100644
index 0000000..8b329a5
--- /dev/null
+++ b/tests/migrations/db_unsafe/initial/default.esdl
@@ -0,0 +1,9 @@
+module default {
+    type Type1 {
+        property always_integer -> int32;
+        property string_int -> str;
+        property int_dur -> int32;
+        property opt_req -> int32;
+        multi property multi_single -> str;
+    }
+}
diff --git a/tests/migrations/db_unsafe/modified1/default.esdl b/tests/migrations/db_unsafe/modified1/default.esdl
new file mode 100644
index 0000000..641a598
--- /dev/null
+++ b/tests/migrations/db_unsafe/modified1/default.esdl
@@ -0,0 +1,14 @@
+module default {
+    type Type1 {
+        # implicit cast
+        property always_integer -> int64;
+        # existing cast
+        property string_int -> int32;
+        # no cast (drop, create)
+        property int_dur -> duration;
+        # make required
+        required property opt_req -> int32;
+        # make single
+        property multi_single -> str;
+    }
+}