// This code is automatically generated by Senax and is always overwritten.
// Senax v@{ ""|senax_version }@

use crate::connection::{DbConn, DbType};
use ::anyhow::Result;
use ::futures::TryStreamExt;
use ::senax_common::ShardId;
use ::std::collections::BTreeMap;
use ::std::path::Path;
use ::std::sync::Arc;
use ::tokio::sync::{Mutex, Semaphore};

pub use _base::models::USE_FAST_CACHE;
pub use _base::models::CACHE_UPDATE_LOCK;
@{-"\n"}@
@%- for (name, defs) in groups %@
pub use _repo_@{ name|snake }@::repositories::@{ name|snake|ident }@;
@%- endfor %@

pub use _base::models::NotifyOp;
pub use _base::models::TableName;
pub use _base::models::CacheHandler;

pub(crate) async fn start(db_dir: &Path) -> Result<()> {
@%- for name in unified_joinable %@
    _base_repo_@{ name }@::start(db_dir).await?;
@%- endfor %@
    Ok(())
}

pub(crate) async fn start_test() -> Result<()> {
@%- for name in unified_joinable %@
    _base_repo_@{ name }@::start_test().await?;
@%- endfor %@
    Ok(())
}

#[rustfmt::skip]
pub(crate) async fn check() -> Result<()> {
    for shard_id in DbConn::shard_num_range() {
        let mut has_error = false;

        let _results: Vec<()> = _check(shard_id).await
            .into_iter()
            .flatten()
            .flatten()
            .filter_map(|r| match r {
                Ok(v) => Some(v),
                Err(e) => {
                    has_error = true;
                    eprintln!("{:?}\n", e);
                    None
                }
            })
            .collect();
        if has_error {
            ::anyhow::bail!("check error!")
        }
    }
    Ok(())
}

#[rustfmt::skip]
async fn _check(shard_id: ShardId) -> Vec<Vec<Vec<Result<()>>>> {
    vec![
        @%- for name in unified_joinable %@
        _base_repo_@{ name }@::check(shard_id).await,
        @%- endfor %@
    ]
}

pub(crate) use _base::models::_clear_cache;

pub(crate) async fn exec_ddl<'c, E>(sql: &str, conn: E) -> Result<()>
where
    E: sqlx::Executor<'c, Database = DbType>,
{
    let mut s = conn.execute_many(sql);
    while s.try_next().await?.is_some() {}
    Ok(())
}

pub(crate) async fn exec_migrate(
    shard_id: ShardId,
    ignore_missing: bool,
    remove_missing: bool,
) -> Result<()> {
    static MIGRATE_LOCK: Mutex<BTreeMap<String, Arc<Semaphore>>> =
        Mutex::const_new(BTreeMap::new());
    let _lock = {
        let mut lock = MIGRATE_LOCK.lock().await;
        lock.entry(DbConn::get_host_name(shard_id).await?)
            .or_insert_with(|| Arc::new(Semaphore::new(1)))
            .clone()
            .acquire_owned()
            .await?
    };
    let conn = DbConn::_new(shard_id);
    let mut writer = conn.acquire_writer().await?;
    @%- if config.is_mysql() %@
    exec_ddl(
        r#"
            CREATE TABLE IF NOT EXISTS _sqlx_migrations (
                version BIGINT PRIMARY KEY,
                description TEXT NOT NULL,
                installed_on DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
                success BOOLEAN NOT NULL,
                checksum BLOB NOT NULL,
                execution_time BIGINT NOT NULL
            );
        "#,
        writer.as_mut(),
    )
    .await?;
    @%- endif %@
    @%- if config.use_sequence || !config.force_disable_cache %@
    @%- if config.is_mysql() %@
    exec_ddl(
        r#"
            CREATE TABLE IF NOT EXISTS _sequence (
                id INT UNSIGNED NOT NULL PRIMARY KEY,
                seq BIGINT UNSIGNED NOT NULL
            );
            INSERT IGNORE INTO _sequence VALUES (1, 0);
            INSERT IGNORE INTO _sequence VALUES (2, 0);
        "#,
        writer.as_mut(),
    )
    .await?;
    @%- else %@
    exec_ddl(
        r#"
            CREATE TABLE IF NOT EXISTS "_sequence" (
                "id" INT NOT NULL PRIMARY KEY,
                "seq" BIGINT NOT NULL
            );
            INSERT INTO "_sequence" VALUES (1, 0) ON CONFLICT DO NOTHING;
            INSERT INTO "_sequence" VALUES (2, 0) ON CONFLICT DO NOTHING;
        "#,
        writer.as_mut(),
    )
    .await?;
    @%- endif %@
    @%- endif %@
    loop {
        match sqlx::migrate!()
            .set_ignore_missing(ignore_missing)
            .run(writer.as_mut())
            .await
        {
            Ok(_) => break,
            Err(sqlx::migrate::MigrateError::VersionMissing(v)) => {
                if remove_missing {
                    exec_ddl(
                        &format!("delete from _sqlx_migrations where version = {};", v),
                        writer.as_mut(),
                    )
                    .await?;
                } else {
                    return Err(sqlx::migrate::MigrateError::VersionMissing(v).into());
                }
            }
            Err(e) => {
                return Err(e.into());
            }
        }
    }
    Ok(())
}
@{-"\n"}@