indexer: DB reset to handle incomplete migration history (#19019)

## Description 

previous fix did not work well when migration run history in
__diesel_schema_migrations table is not complete.

I also considered dropping the database as we usually do locally, but a
connection cannot drop the database in its own connection and ended up
having errors of
```
cannot drop the currently open database
```

thus this pr drops all tables, functions and procedures explicitly for
PG and tables for MySQL as we now only have tables in MySQL.

## Test plan 

1. diesel reset on latest main
2. cherry-pick this pr to 1.31 and 1.30, run cmd below and make sure
that DB are indeed reset
```
DB_POOL_SIZE=10 cargo run --bin sui-indexer -- --db-url =DB_URL --reset-db
```

---

## Release notes

Check each box that your changes affect. If none of the boxes relate to
your changes, release notes aren't required.

For each box you select, include information after the relevant heading
that describes the impact of your changes that a user might notice and
any actions they must take to implement updates.

- [ ] Protocol: 
- [ ] Nodes (Validators and Full nodes): 
- [ ] Indexer: 
- [ ] JSON-RPC: 
- [ ] GraphQL: 
- [ ] CLI: 
- [ ] Rust SDK:
- [ ] REST API:

crates/sui-indexer/src/db.rs

@@ -200,6 +200,7 @@ pub mod setup_postgres {
     use anyhow::anyhow;
     use diesel::migration::MigrationSource;
     use diesel::PgConnection;
+    use diesel::RunQueryDsl;
     use diesel_migrations::{embed_migrations, EmbeddedMigrations, MigrationHarness};
     use prometheus::Registry;
     use secrecy::ExposeSecret;
@@ -208,9 +209,58 @@ pub mod setup_postgres {
     const MIGRATIONS: EmbeddedMigrations = embed_migrations!("migrations/pg");
 
     pub fn reset_database(conn: &mut PoolConnection<PgConnection>) -> Result<(), anyhow::Error> {
-        info!("Resetting database ...");
-        conn.revert_all_migrations(MIGRATIONS)
-            .map_err(|e| anyhow!("Error reverting all migrations {e}"))?;
+        info!("Resetting PG database ...");
+
+        let drop_all_tables = "
+        DO $$ DECLARE
+            r RECORD;
+        BEGIN
+        FOR r IN (SELECT tablename FROM pg_tables WHERE schemaname = 'public')
+            LOOP
+                EXECUTE 'DROP TABLE IF EXISTS ' || quote_ident(r.tablename) || ' CASCADE';
+            END LOOP;
+        END $$;";
+        diesel::sql_query(drop_all_tables).execute(conn)?;
+        info!("Dropped all tables.");
+
+        let drop_all_procedures = "
+        DO $$ DECLARE
+            r RECORD;
+        BEGIN
+            FOR r IN (SELECT proname, oidvectortypes(proargtypes) as argtypes
+                      FROM pg_proc INNER JOIN pg_namespace ns ON (pg_proc.pronamespace = ns.oid)
+                      WHERE ns.nspname = 'public' AND prokind = 'p')
+            LOOP
+                EXECUTE 'DROP PROCEDURE IF EXISTS ' || quote_ident(r.proname) || '(' || r.argtypes || ') CASCADE';
+            END LOOP;
+        END $$;";
+        diesel::sql_query(drop_all_procedures).execute(conn)?;
+        info!("Dropped all procedures.");
+
+        let drop_all_functions = "
+        DO $$ DECLARE
+            r RECORD;
+        BEGIN
+            FOR r IN (SELECT proname, oidvectortypes(proargtypes) as argtypes
+                      FROM pg_proc INNER JOIN pg_namespace ON (pg_proc.pronamespace = pg_namespace.oid)
+                      WHERE pg_namespace.nspname = 'public' AND prokind = 'f')
+            LOOP
+                EXECUTE 'DROP FUNCTION IF EXISTS ' || quote_ident(r.proname) || '(' || r.argtypes || ') CASCADE';
+            END LOOP;
+        END $$;";
+        diesel::sql_query(drop_all_functions).execute(conn)?;
+        info!("Dropped all functions.");
+
+        diesel::sql_query(
+            "
+        CREATE TABLE IF NOT EXISTS __diesel_schema_migrations (
+            version VARCHAR(50) PRIMARY KEY,
+            run_on TIMESTAMP NOT NULL DEFAULT NOW()
+        )",
+        )
+        .execute(conn)?;
+        info!("Created __diesel_schema_migrations table.");
+
         conn.run_migrations(&MIGRATIONS.migrations().unwrap())
             .map_err(|e| anyhow!("Failed to run migrations {e}"))?;
         info!("Reset database complete.");
@@ -307,6 +357,7 @@ pub mod setup_mysql {
     use anyhow::anyhow;
     use diesel::migration::MigrationSource;
     use diesel::MysqlConnection;
+    use diesel::RunQueryDsl;
     use diesel_migrations::{embed_migrations, EmbeddedMigrations, MigrationHarness};
     use prometheus::Registry;
     use secrecy::ExposeSecret;
@@ -315,12 +366,32 @@ pub mod setup_mysql {
     const MIGRATIONS: EmbeddedMigrations = embed_migrations!("migrations/mysql");
 
     pub fn reset_database(conn: &mut PoolConnection<MysqlConnection>) -> Result<(), anyhow::Error> {
-        info!("Resetting database ...");
-        conn.revert_all_migrations(MIGRATIONS)
-            .map_err(|e| anyhow!("Error reverting all migrations {e}"))?;
+        info!("Resetting MySQL database ...");
+
+        let table_names: Vec<String> = diesel::dsl::sql::<diesel::sql_types::Text>(
+            "SELECT TABLE_NAME FROM information_schema.tables WHERE table_schema = DATABASE()",
+        )
+        .load(conn)?;
+        for table_name in table_names {
+            let drop_table_query = format!("DROP TABLE IF EXISTS {}", table_name);
+            diesel::sql_query(drop_table_query).execute(conn)?;
+        }
+        info!("Drop tables complete.");
+
+        diesel::sql_query(
+            "
+            CREATE TABLE __diesel_schema_migrations (
+                version VARCHAR(50) PRIMARY KEY,
+                run_on TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP()
+            )
+        ",
+        )
+        .execute(conn)?;
+        info!("Created __diesel_schema_migrations table.");
+
         conn.run_migrations(&MIGRATIONS.migrations().unwrap())
             .map_err(|e| anyhow!("Failed to run migrations {e}"))?;
-        info!("Reset database complete.");
+        info!("All migrations complete, reset database complete");
         Ok(())
     }