// This code is automatically generated by Senax and is always overwritten.

use anyhow::{ensure, Context as _, Result};
#[allow(unused_imports)]
use crossbeam::queue::SegQueue;
use futures::future::BoxFuture;
use fxhash::{FxHashMap, FxHashSet};
use log::LevelFilter;
#[allow(unused_imports)]
use once_cell::sync::{Lazy, OnceCell};
use senax_common::ShardId;
use sqlx::migrate::MigrateDatabase;
use sqlx::mysql::{MySqlConnectOptions, MySqlPool};
use sqlx::pool::PoolConnection;
use sqlx::{ConnectOptions, Executor, MySqlConnection, Row, Transaction};
use std::collections::hash_map::Entry;
use std::collections::{BTreeMap, VecDeque};
use std::net::SocketAddr;
#[allow(unused_imports)]
use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, AtomicUsize, Ordering};
#[allow(unused_imports)]
use std::sync::Arc;
use std::time::SystemTime;
use std::{cmp, env};
#[allow(unused_imports)]
use tokio::sync::{Mutex, RwLock, Semaphore};
use tokio::task::JoinSet;

use crate::models::{CacheOp, NotifyOp, TableName};
use crate::*;

pub type DbType = sqlx::MySql;
pub type DbPool = MySqlPool;
pub type DbConnection = MySqlConnection;
pub type DbConnectOptions = MySqlConnectOptions;
pub type DbArguments = sqlx::mysql::MySqlArguments;
pub type DbRow = sqlx::mysql::MySqlRow;
pub const TX_ISOLATION: Option<&'static str> = @{ tx_isolation|disp_opt }@;
pub const READ_TX_ISOLATION: Option<&'static str> = @{ read_tx_isolation|disp_opt }@;
const CHECK_SQL: &str = "select @@innodb_read_only OR @@read_only OR @@super_read_only";
const CHECK_SQL_WRITABLE_RESULT: i8 = 0;
@%- if config.use_sequence %@
const ID_OF_SEQUENCE: u32 = 1;
@%- endif %@
@%- if !config.force_disable_cache %@
const ID_OF_CACHE_SYNC: u32 = 2;
@%- endif %@
const DISABLE_CACHE: bool = @{ config.force_disable_cache }@ || cfg!(feature = "cache_update_only");

static DB_URL: RwLock<String> = RwLock::const_new(String::new());
static DB_USER: RwLock<Option<String>> = RwLock::const_new(None);
static DB_PASSWORD: RwLock<Option<String>> = RwLock::const_new(None);
static REAL_DB_NAME: OnceCell<Vec<String>> = OnceCell::new();
static REPLICA_DB_URL: RwLock<String> = RwLock::const_new(String::new());
static REPLICA_DB_USER: RwLock<Option<String>> = RwLock::const_new(None);
static REPLICA_DB_PASSWORD: RwLock<Option<String>> = RwLock::const_new(None);
static CACHE_DB_USE_SOURCE: AtomicBool = AtomicBool::new(false);
static MAX_CONNECTIONS_FOR_WRITE: AtomicU32 = AtomicU32::new(0);
static MAX_CONNECTIONS_FOR_READ: AtomicU32 = AtomicU32::new(0);
static MAX_CONNECTIONS_FOR_CACHE: AtomicU32 = AtomicU32::new(0);
static SEQUENCE_FETCH_NUM: AtomicU32 = AtomicU32::new(0);
static ENABLE_FAST_FAILOVER: AtomicBool = AtomicBool::new(false);
static USE_IPV4_ONLY: AtomicBool = AtomicBool::new(false);
static USE_IPV6_ONLY: AtomicBool = AtomicBool::new(false);

static WRITER: RwLock<Vec<Option<(SocketAddr, DbPool)>>> = RwLock::const_new(Vec::new());
static SHARD_NUM: OnceCell<usize> = OnceCell::new();
type AddrPoolMap = FxHashMap<SocketAddr, (Arc<PoolInfo>, Option<bool>)>;
static READER: OnceCell<Vec<Arc<RwLock<AddrPoolMap>>>> = OnceCell::new();
static CACHE: OnceCell<Vec<Arc<RwLock<AddrPoolMap>>>> = OnceCell::new();
@%- if config.use_sequence %@
static SEQUENCE: Lazy<Vec<Mutex<(u64, u64)>>> = Lazy::new(|| {
    DbConn::shard_num_range()
        .map(|_| Mutex::new((0, 0)))
        .collect()
});
@%- endif %@
type NotifyFn = Box<dyn Fn(TableName, NotifyOp, &str) + Send + Sync>;
static NOTIFY_RECEIVER: RwLock<Vec<NotifyFn>> = RwLock::const_new(Vec::new());
static NOTIFY_RECEIVER_COUNT: AtomicUsize = AtomicUsize::new(0);
static NOTIFY_LIST: RwLock<Option<FxHashMap<(TableName, String), NotifyOp>>> =
    RwLock::const_new(None);

struct PoolInfo {
    addr: SocketAddr,
    target: Target,
    inner: DbPool,
}

impl Drop for PoolInfo {
    fn drop(&mut self) {
        log::info!("remove connection for {}: {}", self.target, self.addr);
    }
}

#[derive(PartialEq, Eq, Clone, Copy, strum::Display)]
enum Target {
    #[strum(to_string = "write")]
    Write,
    #[strum(to_string = "read")]
    Read,
    #[strum(to_string = "cache")]
    Cache,
}

macro_rules! split_shard {
    ( $x:expr ) => {{
        $x.split('\n').map(|v| v.trim()).filter(|v| !v.is_empty())
    }};
}

#[rustfmt::skip]
async fn config(etcd: &FxHashMap<String, String>, test_mode: bool) -> Result<()> {
    if test_mode {
        let s_url = env_urls(etcd, "@{ db|upper_snake }@_TEST_DB_URL")?
            .with_context(|| "@{ db|upper_snake }@_TEST_DB_URL is required in the .env file.")?;
        let r_url = s_url.clone();
        *DB_URL.write().await = s_url;
        *REPLICA_DB_URL.write().await = r_url;
        let s_user = env_opt_str(etcd, "@{ db|upper_snake }@_TEST_DB_USER");
        let r_user = s_user.clone();
        *DB_USER.write().await = s_user;
        *REPLICA_DB_USER.write().await = r_user;
        let s_pw = env_opt_str(etcd, "@{ db|upper_snake }@_TEST_DB_PASSWORD");
        let r_pw = s_pw.clone();
        *DB_PASSWORD.write().await = s_pw;
        *REPLICA_DB_PASSWORD.write().await = r_pw;
    } else {
        let s_url = env_urls(etcd, "@{ db|upper_snake }@_DB_URL")?
            .with_context(|| "@{ db|upper_snake }@_DB_URL is required in the .env file.")?;
        let r_url = env_urls(etcd, "@{ db|upper_snake }@_REPLICA_DB_URL")?.unwrap_or_else(|| s_url.clone());
        *DB_URL.write().await = s_url;
        *REPLICA_DB_URL.write().await = r_url;
        let s_user = env_opt_str(etcd, "@{ db|upper_snake }@_DB_USER");
        let r_user = env_opt_str(etcd, "@{ db|upper_snake }@_REPLICA_DB_USER").or_else(|| s_user.clone());
        *DB_USER.write().await = s_user;
        *REPLICA_DB_USER.write().await = r_user;
        let s_pw = env_opt_str(etcd, "@{ db|upper_snake }@_DB_PASSWORD");
        let r_pw = env_opt_str(etcd, "@{ db|upper_snake }@_REPLICA_DB_PASSWORD").or_else(|| s_pw.clone());
        *DB_PASSWORD.write().await = s_pw;
        *REPLICA_DB_PASSWORD.write().await = r_pw;
        let cache_use_source = env_opt_str(etcd, "@{ db|upper_snake }@_CACHE_DB_USE_SOURCE")
            .map(|v| v.parse())
            .transpose()?
            .unwrap_or_default();
        CACHE_DB_USE_SOURCE.store(cache_use_source, Ordering::SeqCst);
    }
    if SHARD_NUM.get().is_none() {
        let s_url = DB_URL.read().await.to_owned();
        let mut v = Vec::new();
        for source in split_shard!(s_url) {
            let url = url::Url::parse(source)?;
            v.push(format!("\"{}\".", url.path().trim_matches('/')));
        }
        let _ = REAL_DB_NAME.set(v);
        let r_url = REPLICA_DB_URL.read().await.to_owned();
        let source_len = split_shard!(s_url).count();
        ensure!(
            source_len <= ShardId::MAX as usize + 1,
            "Number of shards exceeds limit."
        );
        let replica_len = split_shard!(r_url).count();
        ensure!(
            source_len == replica_len,
            "Number of shards for source and replica are different."
        );
        let _ = SHARD_NUM.set(source_len);
        *WRITER.write().await = vec![None; source_len];
        let _ = READER.set(
            std::iter::repeat_with(|| Arc::new(RwLock::new(FxHashMap::default())))
                .take(source_len)
                .collect(),
        );
        let _ = CACHE.set(
            std::iter::repeat_with(|| Arc::new(RwLock::new(FxHashMap::default())))
                .take(source_len)
                .collect(),
        );
    }

    let v = env_u32(
        etcd,
        "@{ db|upper_snake }@_DB_MAX_CONNECTIONS_FOR_WRITE",
        DEFAULT_DB_MAX_CONNECTIONS_FOR_WRITE,
    )?;
    MAX_CONNECTIONS_FOR_WRITE.store(v, Ordering::SeqCst);
    let v = env_u32(
        etcd,
        "@{ db|upper_snake }@_DB_MAX_CONNECTIONS_FOR_READ",
        DEFAULT_DB_MAX_CONNECTIONS_FOR_READ,
    )?;
    MAX_CONNECTIONS_FOR_READ.store(v, Ordering::SeqCst);
    let v = env_u32(
        etcd,
        "@{ db|upper_snake }@_DB_MAX_CONNECTIONS_FOR_CACHE",
        DEFAULT_DB_MAX_CONNECTIONS_FOR_CACHE,
    )?;
    MAX_CONNECTIONS_FOR_CACHE.store(v, Ordering::SeqCst);
    let v = env_u32(etcd, "@{ db|upper_snake }@_SEQUENCE_FETCH_NUM", DEFAULT_SEQUENCE_FETCH_NUM)?;
    SEQUENCE_FETCH_NUM.store(cmp::max(1, v), Ordering::SeqCst);
    if let Some(v) = env_opt_str(etcd, "@{ db|upper_snake }@_ENABLE_FAST_FAILOVER") {
        ENABLE_FAST_FAILOVER.store(v.parse()?, Ordering::SeqCst);
    }
    if let Some(v) = env_opt_str(etcd, "@{ db|upper_snake }@_USE_IPV4_ONLY") {
        USE_IPV4_ONLY.store(v.parse()?, Ordering::SeqCst);
    }
    if let Some(v) = env_opt_str(etcd, "@{ db|upper_snake }@_USE_IPV6_ONLY") {
        USE_IPV6_ONLY.store(v.parse()?, Ordering::SeqCst);
    }
    Ok(())
}

pub async fn init() -> Result<()> {
    log::info!("Connecting to database");
    #[cfg(feature = "etcd")]
    let etcd = senax_common::etcd::map("db/").await?;
    #[cfg(not(feature = "etcd"))]
    let etcd = FxHashMap::default();
    config(&etcd, false).await?;
    connect(true, true, true, None).await?;

    let writer = WRITER.read().await;
    ensure!(!writer.is_empty(), "There are no writable database shards.");
    for (i, s) in writer.iter().enumerate() {
        ensure!(
            s.is_some(),
            "There are no writable database connections.(code:0, {i})"
        );
    }

    let reader = READER.get().unwrap();
    ensure!(!reader.is_empty(), "There are no readable database shards.");
    for (i, s) in reader.iter().enumerate() {
        let s = s.read().await;
        ensure!(
            !s.is_empty(),
            "There are no readable database connections.(code:0, {i})"
        );
    }

    if !DISABLE_CACHE {
        let cache = CACHE.get().unwrap();
        ensure!(!cache.is_empty(), "There are no database shards for cache.");
        for (i, s) in cache.iter().enumerate() {
            let s = s.read().await;
            ensure!(
                !s.is_empty(),
                "There are no database connections for cache.(code:0, {i})"
            );
        }
    }
    log::info!("Connection to database completed.");

    tokio::spawn(async {
        loop {
            tokio::time::sleep(Duration::from_secs(crate::CONNECT_CHECK_INTERVAL)).await;
            reconnect().await;
        }
    });

    #[cfg(feature = "etcd")]
    tokio::spawn(async {
        loop {
            let mut stream = match senax_common::etcd::watch("db/", true).await {
                Ok(stream) => stream,
                Err(err) => {
                    log::error!("{}", err);
                    tokio::time::sleep(Duration::from_secs(60)).await;
                    continue;
                }
            };
            loop {
                match stream.message().await {
                    Ok(Some(_)) => {}
                    Ok(None) => {
                        log::error!("etcd connection has been lost.");
                        break;
                    }
                    Err(err) => {
                        log::error!("{}", err);
                        break;
                    }
                }
                let etcd = senax_common::etcd::map("db/").await?;
                config(&etcd, false).await?;
                reconnect().await;
            }
        }
    });
    Ok(())
}

pub async fn reconnect() {
    let mut join_set = JoinSet::new();
    for shard_id in DbConn::shard_num_range() {
        join_set.spawn(async move {
            if let Err(e) = connect(true, false, false, Some(shard_id)).await {
                log::error!("{}", e);
            }
            tokio::time::sleep(Duration::from_millis(100)).await;
            if let Err(e) = connect(false, true, false, Some(shard_id)).await {
                log::error!("{}", e);
            }
            tokio::time::sleep(Duration::from_millis(100)).await;
            if let Err(e) = connect(false, false, true, Some(shard_id)).await {
                log::error!("{}", e);
            }
        });
    }
    while let Some(r) = join_set.join_next().await {
        if let Err(e) = r {
            log::error!("{}", e);
        }
    }
}

pub async fn connect(
    update_writer: bool,
    update_reader: bool,
    update_cache: bool,
    shard_id: Option<ShardId>,
) -> Result<()> {
    static WRITER_LOCK: Lazy<Vec<Semaphore>> = Lazy::new(|| {
        std::iter::repeat_with(|| Semaphore::new(1))
            .take(DbConn::shard_num())
            .collect()
    });
    static READER_LOCK: Lazy<Vec<Semaphore>> = Lazy::new(|| {
        std::iter::repeat_with(|| Semaphore::new(1))
            .take(DbConn::shard_num())
            .collect()
    });
    static CACHE_LOCK: Lazy<Vec<Semaphore>> = Lazy::new(|| {
        std::iter::repeat_with(|| Semaphore::new(1))
            .take(DbConn::shard_num())
            .collect()
    });

    let cache_use_source = CACHE_DB_USE_SOURCE.load(Ordering::Relaxed);
    let update_source = update_writer || (cache_use_source && update_cache);
    let update_replica = update_reader || (!cache_use_source && update_cache);
    let update_writer = update_source;
    let update_reader = update_replica;
    let update_cache = (cache_use_source && update_source) || (!cache_use_source && update_replica);

    let mut join_set = JoinSet::new();
    if update_writer {
        join_set.spawn(async move {
            tokio::spawn(async move {
                let _lock = if let Some(shard_id) = shard_id {
                    if let Some(lock) = WRITER_LOCK.get(shard_id as usize) {
                        match lock.try_acquire() {
                            Ok(lock) => Some(lock),
                            Err(_) => {
                                let _ = lock.acquire().await;
                                return Ok(());
                            }
                        }
                    } else {
                        return Ok(());
                    }
                } else {
                    None
                };

                let s_url = DB_URL.read().await.to_owned();
                let s_user = DB_USER.read().await.to_owned();
                let s_pw = DB_PASSWORD.read().await.to_owned();
                let r_url = REPLICA_DB_URL.read().await.to_owned();
                let r_user = REPLICA_DB_USER.read().await.to_owned();
                let r_pw = REPLICA_DB_PASSWORD.read().await.to_owned();

                reset_writer_pool(
                    &s_url,
                    &s_user,
                    &s_pw,
                    &r_url,
                    &r_user,
                    &r_pw,
                    db_options_for_write(),
                    &WRITER,
                    shard_id,
                )
                .await
            })
            .await
        });
    }

    if update_reader {
        join_set.spawn(async move {
            tokio::spawn(async move {
                let _lock = if let Some(shard_id) = shard_id {
                    if let Some(lock) = READER_LOCK.get(shard_id as usize) {
                        match lock.try_acquire() {
                            Ok(lock) => Some(lock),
                            Err(_) => {
                                let _ = lock.acquire().await;
                                return Ok(());
                            }
                        }
                    } else {
                        return Ok(());
                    }
                } else {
                    None
                };

                let s_url = DB_URL.read().await.to_owned();
                let s_user = DB_USER.read().await.to_owned();
                let s_pw = DB_PASSWORD.read().await.to_owned();
                let r_url = REPLICA_DB_URL.read().await.to_owned();
                let r_user = REPLICA_DB_USER.read().await.to_owned();
                let r_pw = REPLICA_DB_PASSWORD.read().await.to_owned();

                reset_reader_pool(
                    &r_url,
                    &r_user,
                    &r_pw,
                    &s_url,
                    &s_user,
                    &s_pw,
                    db_options_for_read(),
                    &READER,
                    shard_id,
                    Target::Read,
                )
                .await
            })
            .await
        });
    }

    if update_cache && !DISABLE_CACHE {
        join_set.spawn(async move {
            tokio::spawn(async move {
                let _lock = if let Some(shard_id) = shard_id {
                    if let Some(lock) = CACHE_LOCK.get(shard_id as usize) {
                        match lock.try_acquire() {
                            Ok(lock) => Some(lock),
                            Err(_) => {
                                let _ = lock.acquire().await;
                                return Ok(());
                            }
                        }
                    } else {
                        return Ok(());
                    }
                } else {
                    None
                };

                let s_url = DB_URL.read().await.to_owned();
                let s_user = DB_USER.read().await.to_owned();
                let s_pw = DB_PASSWORD.read().await.to_owned();
                let r_url = REPLICA_DB_URL.read().await.to_owned();
                let r_user = REPLICA_DB_USER.read().await.to_owned();
                let r_pw = REPLICA_DB_PASSWORD.read().await.to_owned();

                let (r_url, r_user, r_pw, s_url, s_user, s_pw) = if cache_use_source {
                    (s_url, s_user, s_pw, r_url, r_user, r_pw)
                } else {
                    (r_url, r_user, r_pw, s_url, s_user, s_pw)
                };

                reset_reader_pool(
                    &r_url,
                    &r_user,
                    &r_pw,
                    &s_url,
                    &s_user,
                    &s_pw,
                    db_options_for_cache(),
                    &CACHE,
                    shard_id,
                    Target::Cache,
                )
                .await
            })
            .await
        });
    }
    while let Some(r) = join_set.join_next().await {
        r???;
    }
    Ok(())
}

#[rustfmt::skip]
pub async fn init_test() -> Result<()> {
    let source_len = DbConn::shard_num();
    *WRITER.write().await = vec![None; source_len];
    if let Some(list) = READER.get() {
        for pool in list {
            pool.write().await.clear();
        }
    }
    if let Some(list) = CACHE.get() {
        for pool in list {
            pool.write().await.clear();
        }
    }
    let etcd = FxHashMap::default();
    config(&etcd, true).await?;
    connect(true, true, true, None).await?;
    Ok(())
}

#[rustfmt::skip]
pub(crate) async fn reset_database(is_test: bool, clean: bool) -> Result<()> {
    #[cfg(feature = "etcd")]
    let etcd = senax_common::etcd::map("db/").await?;
    #[cfg(not(feature = "etcd"))]
    let etcd = FxHashMap::default();
    config(&etcd, is_test).await?;
    let db_url = DB_URL.read().await.to_owned();
    let user = DB_USER.read().await.to_owned();
    let pw = DB_PASSWORD.read().await.to_owned();

    for url in db_url.split('\n') {
        let url = url.trim();
        if url.is_empty() {
            continue;
        }
        let mut url: url::Url = url.parse()?;
        if let Some(user) = &user {
            url.set_username(user).expect("DB_URL ERROR");
        }
        if pw.is_some() {
            url.set_password(pw.as_deref()).expect("DB_URL ERROR");
        }
        let url = url.as_str();
        if clean {
            DbType::drop_database(url).await?;
            DbType::create_database(url).await?;
        } else if !DbType::database_exists(url).await? {
            DbType::create_database(url).await?;
        }
    }
    Ok(())
}

struct SavePoint {
    cache_internal_op_pos: usize,
    cache_op_pos: usize,
    callback_post: usize,
}
pub struct DbConn {
    ctx_no: u64,
    time: SystemTime,
    shard_id: ShardId,
    tx: FxHashMap<ShardId, Transaction<'static, DbType>>,
    save_point: Vec<SavePoint>,
    read_tx: FxHashMap<ShardId, Transaction<'static, DbType>>,
    @%- if !config.force_disable_cache %@
    cache_tx: FxHashMap<ShardId, (u64, Transaction<'static, DbType>)>,
    @%- endif %@
    conn: FxHashMap<ShardId, PoolConnection<DbType>>,
    cache_internal_op_list: Vec<(ShardId, CacheOp)>,
    cache_op_list: Vec<(ShardId, CacheOp)>,
    callback_list: VecDeque<Box<dyn FnOnce() -> BoxFuture<'static, ()> + Send + Sync>>,
    pub(crate) clear_whole_cache: bool,
    has_tx: bool,
    wo_tx: usize,
    has_read_tx: usize,
    lock_list: Vec<DbLock>,
    @%- for db in config.outer_db() %@
    pub(crate) _@{ db }@_db: ::db_@{ db }@::connection::DbConn,
    @%- endfor %@
}

impl Clone for DbConn {
    fn clone(&self) -> Self {
        DbConn::__new(self.ctx_no, self.time, self.shard_id)
    }
}

impl DbConn {
    pub fn new(ctx_no: u64) -> DbConn {
        DbConn::__new(ctx_no, SystemTime::now(), 0)
    }

    pub fn new_with_time(ctx_no: u64, time: SystemTime) -> DbConn {
        DbConn::__new(ctx_no, time, 0)
    }

    pub(crate) fn _new(shard_id: ShardId) -> DbConn {
        DbConn::__new(0, SystemTime::now(), shard_id)
    }

    pub(crate) fn _new_with_ctx(ctx_no: u64, shard_id: ShardId) -> DbConn {
        DbConn::__new(ctx_no, SystemTime::now(), shard_id)
    }

    pub fn __new(ctx_no: u64, time: SystemTime, shard_id: ShardId) -> DbConn {
        DbConn {
            ctx_no,
            time,
            shard_id: shard_id % (Self::shard_num() as ShardId),
            tx: FxHashMap::default(),
            save_point: Vec::new(),
            read_tx: FxHashMap::default(),
            @%- if !config.force_disable_cache %@
            cache_tx: FxHashMap::default(),
            @%- endif %@
            conn: FxHashMap::default(),
            cache_internal_op_list: Vec::new(),
            cache_op_list: Vec::new(),
            callback_list: VecDeque::new(),
            clear_whole_cache: false,
            has_tx: false,
            wo_tx: 0,
            has_read_tx: 0,
            lock_list: Vec::new(),
            @%- for db in config.outer_db() %@
            _@{ db }@_db: ::db_@{ db }@::connection::DbConn::__new(ctx_no, time, shard_id),
            @%- endfor %@
        }
    }

    pub fn ctx_no(&self) -> u64 {
        self.ctx_no
    }

    pub fn real_db_name(shard_id: ShardId) -> &'static str {
        &REAL_DB_NAME.get().unwrap()[shard_id as usize % DbConn::shard_num()]
    }

    pub async fn get_host_name(shard_id: ShardId) -> Result<String> {
        let s_url = DB_URL.read().await;
        let url = split_shard!(s_url)
            .nth(shard_id as usize)
            .ok_or_else(|| anyhow::anyhow!("shard_id is out of range."))?;
        let url = url::Url::parse(url)?;
        Ok(url.host_str().unwrap().to_string())
    }
    @%- for db in config.outer_db() %@

    pub fn _@{ db }@_db(&mut self) -> &mut ::db_@{ db }@::connection::DbConn {
        &mut self._@{ db }@_db
    }
    @%- endfor %@

    pub fn set_time(&mut self, time: SystemTime) {
        self.time = time;
    }

    pub fn time(&self) -> SystemTime {
        self.time
    }

    pub fn shard_num() -> usize {
        *SHARD_NUM.get().unwrap()
    }

    pub fn shard_num_range() -> std::ops::RangeInclusive<ShardId> {
        0..=(Self::shard_num() - 1) as ShardId
    }

    pub fn shard_id(&self) -> ShardId {
        self.shard_id
    }

    pub fn set_shard_id(&mut self, shard_id: usize) {
        self.shard_id = shard_id as ShardId;
    }
    @%- if !config.force_disable_cache %@

    /// Cache transaction synchronization
    #[allow(dead_code)]
    pub(crate) fn cache_sync(&self) -> u64 {
        self.cache_tx.get(&self.shard_id).map(|v| v.0).unwrap_or(0)
    }

    pub fn set_clear_all_cache(&mut self) {
        self.clear_whole_cache = true;
    }
    @%- endif %@

    async fn __acquire_writer(shard_id: ShardId) -> Result<PoolConnection<DbType>> {
        ensure!(
            Self::shard_num_range().contains(&shard_id),
            "shard_id is out of range."
        );
        let (_, pool) = WRITER
            .read()
            .await
            .get(shard_id as usize)
            .context("There are no writable database connections.(code:1)")?
            .clone()
            .context("There are no writable database connections.(code:2)")?;
        let mut conn = pool.acquire().await?;
        let row: (i8,) = sqlx::query_as(CHECK_SQL).fetch_one(conn.as_mut()).await?;
        ensure!(
            row.0 == CHECK_SQL_WRITABLE_RESULT,
            "There are no writable database connections.(code:3)"
        );
        Ok(conn)
    }

    async fn _acquire_writer(shard_id: ShardId) -> Result<PoolConnection<DbType>> {
        if let Ok(conn) = Self::__acquire_writer(shard_id).await {
            Ok(conn)
        } else {
            let now = SystemTime::now();
            if let Err(e) = connect(true, false, false, Some(shard_id)).await {
                log::warn!("{}", e);
            }
            if let Ok(conn) = Self::__acquire_writer(shard_id).await {
                Ok(conn)
            } else {
                if let Some(time) = Duration::from_secs(crate::ACQUIRE_CONNECTION_WAIT_TIME)
                    .checked_sub(now.elapsed().unwrap_or_default())
                {
                    tokio::time::sleep(time).await;
                }
                connect(true, false, false, Some(shard_id)).await?;
                Self::__acquire_writer(shard_id).await
            }
        }
    }

    async fn _acquire_writer_tx(shard_id: ShardId) -> Result<Transaction<'static, DbType>> {
        let conn = Self::_acquire_writer(shard_id).await?;
        Ok(Transaction::begin(conn, None).await?)
    }

    pub async fn acquire_writer(&self) -> Result<PoolConnection<DbType>> {
        Self::_acquire_writer(self.shard_id).await
    }

    async fn acquire_from_pool(
        shard_id: ShardId,
        pool_collection: &OnceCell<Vec<Arc<RwLock<AddrPoolMap>>>>,
    ) -> Result<PoolConnection<DbType>> {
        ensure!(
            Self::shard_num_range().contains(&shard_id),
            "shard_id is out of range."
        );
        let pools: Vec<_> = pool_collection.get().unwrap()[shard_id as usize]
            .read()
            .await
            .values()
            .cloned()
            .collect();
        let len = pools.len();
        ensure!(
            len != 0,
            "There are no readable database connections.(code:11)"
        );
        let mut indexes = Vec::with_capacity(len);
        for i in 0..len {
            indexes.push(i);
        }
        use rand::seq::SliceRandom;
        use rand::thread_rng;
        indexes.shuffle(&mut thread_rng());
        let mut join = JoinSet::new();
        for (i, (pool, _)) in std::iter::zip(indexes, pools) {
            join.spawn(async move {
                tokio::time::sleep(Duration::from_millis(20 * i as u64)).await;
                // spawn to prevent processing interruptions until a newly acquired connection enters the pool
                tokio::spawn(async move { pool.inner.acquire().await }).await
            });
        }
        while let Some(r) = join.join_next().await {
            if let Ok(Ok(Ok(r))) = r {
                return Ok(r);
            }
        }
        anyhow::bail!("There are no readable database connections.(code:12)")
    }

    async fn _acquire_reader(shard_id: ShardId) -> Result<PoolConnection<DbType>> {
        if let Ok(conn) = Self::acquire_from_pool(shard_id, &READER).await {
            Ok(conn)
        } else {
            let now = SystemTime::now();
            if let Err(e) = connect(false, true, false, Some(shard_id)).await {
                log::warn!("{}", e);
            }
            if let Ok(conn) = Self::acquire_from_pool(shard_id, &READER).await {
                Ok(conn)
            } else {
                if let Some(time) = Duration::from_secs(crate::ACQUIRE_CONNECTION_WAIT_TIME)
                    .checked_sub(now.elapsed().unwrap_or_default())
                {
                    tokio::time::sleep(time).await;
                }
                connect(false, true, false, Some(shard_id)).await?;
                Self::acquire_from_pool(shard_id, &READER).await
            }
        }
    }

    pub async fn acquire_reader(&self) -> Result<PoolConnection<DbType>> {
        Self::_acquire_reader(self.shard_id).await
    }

    async fn acquire_reader_tx(shard_id: ShardId) -> Result<Transaction<'static, DbType>> {
        let conn = Self::_acquire_reader(shard_id).await?;
        Ok(Transaction::begin(conn, None).await?)
    }
    @%- if !config.force_disable_cache %@

    async fn __acquire_cache(shard_id: ShardId) -> Result<PoolConnection<DbType>> {
        Self::acquire_from_pool(shard_id, &CACHE).await
    }

    async fn _acquire_cache(shard_id: ShardId) -> Result<PoolConnection<DbType>> {
        if let Ok(conn) = Self::__acquire_cache(shard_id).await {
            Ok(conn)
        } else {
            let now = SystemTime::now();
            if let Err(e) = connect(false, false, true, Some(shard_id)).await {
                log::warn!("{}", e);
            }
            if let Ok(conn) = Self::__acquire_cache(shard_id).await {
                Ok(conn)
            } else {
                if let Some(time) = Duration::from_secs(crate::ACQUIRE_CONNECTION_WAIT_TIME)
                    .checked_sub(now.elapsed().unwrap_or_default())
                {
                    tokio::time::sleep(time).await;
                }
                connect(false, false, true, Some(shard_id)).await?;
                Self::__acquire_cache(shard_id).await
            }
        }
    }

    async fn acquire_cache_tx(shard_id: ShardId) -> Result<Transaction<'static, DbType>> {
        let conn = Self::_acquire_cache(shard_id).await?;
        Ok(Transaction::begin(conn, None).await?)
    }
    @%- endif %@

    pub async fn get_reader(&mut self) -> Result<&mut PoolConnection<DbType>> {
        let conn = if self.conn.contains_key(&self.shard_id) {
            None
        } else {
            Some(self.acquire_reader().await?)
        };
        Ok(self
            .conn
            .entry(self.shard_id)
            .or_insert_with(|| conn.unwrap()))
    }

    pub fn release_conn(&mut self) {
        self.conn.clear();
        @%- for db in config.outer_db() %@
        self._@{ db }@_db.release_conn();
        @%- endfor %@
    }

    pub async fn begin_without_transaction(&mut self) -> Result<()> {
        self.wo_tx += 1;
        @%- for db in config.outer_db() %@
        self._@{ db }@_db.begin_without_transaction().await?;
        @%- endfor %@
        Ok(())
    }

    pub async fn end_of_without_transaction(&mut self) -> Result<()> {
        ensure!(self.wo_tx > 0, "No without transaction is active.");
        self.wo_tx -= 1;
        @%- for db in config.outer_db() %@
        self._@{ db }@_db.end_of_without_transaction().await?;
        @%- endfor %@
        Ok(())
    }

    pub async fn begin(&mut self) -> Result<()> {
        if !self.has_tx {
            self.has_tx = true;
        } else {
            for (_shard_id, tx) in self.tx.iter_mut() {
                let sql = format!("SAVEPOINT s{}", self.save_point.len());
                tx.execute(&*sql).await?;
            }
            self.save_point.push(SavePoint {
                cache_internal_op_pos: self.cache_internal_op_list.len(),
                cache_op_pos: self.cache_op_list.len(),
                callback_post: self.callback_list.len(),
            });
        }
        @%- for db in config.outer_db() %@
        self._@{ db }@_db.begin().await?;
        @%- endfor %@
        Ok(())
    }

    pub async fn begin_immediately(&mut self) -> Result<()> {
        self.begin().await?;
        self.get_tx().await?;
        Ok(())
    }

    pub fn has_tx(&self) -> bool {
        self.has_tx
    }

    pub(crate) fn wo_tx(&self) -> bool {
        !self.has_tx && self.wo_tx > 0
    }

    pub async fn get_tx(&mut self) -> Result<&mut Transaction<'static, DbType>> {
        ensure!(self.has_tx(), "No transaction is active.");
        match self.tx.entry(self.shard_id) {
            Entry::Occupied(tx) => Ok(tx.into_mut()),
            Entry::Vacant(v) => {
                let mut tx = Self::_acquire_writer_tx(self.shard_id).await?;
                set_tx_isolation(&mut tx).await?;
                for s in 0..self.save_point.len() {
                    let sql = format!("SAVEPOINT s{}", s);
                    tx.execute(&*sql).await?;
                }
                Ok(v.insert(tx))
            }
        }
    }

    #[allow(dead_code)]
    pub(crate) async fn push_cache_op(&mut self, op: CacheOp) {
        if self.has_tx() {
            self.cache_op_list.push((self.shard_id(), op));
        } else {
            let shard_id = self.shard_id();
            tokio::spawn(async move {
                let sync = Self::inc_cache_sync(shard_id).await;
                let mut sync_map = FxHashMap::default();
                sync_map.insert(shard_id, sync);
                CacheMsg(vec![op], sync_map).do_send().await;
            });
        }
    }

    #[allow(dead_code)]
    pub(crate) async fn push_cache_op_to(&mut self, op: CacheOp, internal: bool) {
        if internal {
            if self.has_tx() {
                self.cache_internal_op_list.push((self.shard_id(), op));
            } else {
                let shard_id = self.shard_id();
                tokio::spawn(async move {
                    let sync = Self::inc_cache_sync(shard_id).await;
                    let mut sync_map = FxHashMap::default();
                    sync_map.insert(shard_id, sync);
                    CacheMsg(vec![op], sync_map).do_send_to_internal().await;
                });
            }
        } else {
            self.push_cache_op(op).await;
        }
    }

    #[allow(dead_code)]
    pub(crate) async fn push_callback(
        &mut self,
        cb: Box<dyn FnOnce() -> BoxFuture<'static, ()> + Send + Sync>,
    ) {
        if self.has_tx() {
            self.callback_list.push_back(cb);
        } else {
            cb().await;
        }
    }

    pub async fn rows_affected(&mut self) -> Result<i64> {
        let query = sqlx::query("select row_count()");
        let row = query.fetch_one(self.get_tx().await?.as_mut()).await?;
        Ok(row.try_get(0)?)
    }

    pub async fn commit(&mut self) -> Result<()> {
        ensure!(self.has_tx(), "No transaction is active.");
        @%- for db in config.outer_db() %@
        self._@{ db }@_db.commit().await?;
        @%- endfor %@
        if let Some(_save_point) = self.save_point.pop() {
            for (_shard_id, tx) in self.tx.iter_mut() {
                let sql = format!("RELEASE SAVEPOINT s{}", self.save_point.len());
                tx.execute(&*sql).await?;
            }
            return Ok(());
        }
        self.has_tx = false;
        let mut cache_internal_op_list: FxHashMap<ShardId, Vec<CacheOp>> = FxHashMap::default();
        for (s, v) in self.cache_internal_op_list.drain(..) {
            cache_internal_op_list.entry(s).or_default().push(v);
        }
        let mut cache_op_list: FxHashMap<ShardId, Vec<CacheOp>> = FxHashMap::default();
        for (s, v) in self.cache_op_list.drain(..) {
            cache_op_list.entry(s).or_default().push(v);
        }
        let mut join_list: Vec<tokio::task::JoinHandle<Result<()>>> = Vec::new();
        for (shard_id, tx) in self.tx.drain() {
            let cache_internal_op_list = cache_internal_op_list.remove(&shard_id);
            let cache_op_list = cache_op_list.remove(&shard_id);
            let clear_whole_cache = self.clear_whole_cache;
            // JoinSet is not used to prevent processing interruptions
            join_list.push(tokio::spawn(async move {
                tx.commit().await?;
                if !clear_whole_cache
                    && (cache_internal_op_list.is_some() || cache_op_list.is_some())
                {
                    let sync = Self::inc_cache_sync(shard_id).await;
                    let mut sync_map = FxHashMap::default();
                    sync_map.insert(shard_id, sync);
                    if let Some(cache_internal_op_list) = cache_internal_op_list {
                        CacheMsg(cache_internal_op_list, sync_map.clone())
                            .do_send_to_internal()
                            .await;
                    }
                    if let Some(cache_op_list) = cache_op_list {
                        CacheMsg(cache_op_list, sync_map).do_send().await;
                    }
                }
                Ok(())
            }));
        }
        for join in join_list {
            join.await??;
        }
        self.lock_list.clear();
        if self.clear_whole_cache {
            crate::clear_whole_cache().await;
        }
        if !self.callback_list.is_empty() {
            let mut fut = Vec::new();
            while let Some(cb) = self.callback_list.pop_front() {
                fut.push(cb());
            }
            tokio::spawn(async move {
                futures::future::join_all(fut).await;
            });
        }
        Ok(())
    }

    pub async fn rollback(&mut self) -> Result<()> {
        ensure!(self.has_tx(), "No transaction is active.");
        @%- for db in config.outer_db() %@
        self._@{ db }@_db.rollback().await?;
        @%- endfor %@
        if let Some(save_point) = self.save_point.pop() {
            self.cache_internal_op_list
                .truncate(save_point.cache_internal_op_pos);
            self.cache_op_list.truncate(save_point.cache_op_pos);
            self.callback_list.truncate(save_point.callback_post);
            for (_shard_id, tx) in self.tx.iter_mut() {
                let sql = format!("ROLLBACK TO SAVEPOINT s{}", self.save_point.len());
                tx.execute(&*sql).await?;
            }
            return Ok(());
        }
        self.has_tx = false;
        for (_shard_id, tx) in self.tx.drain() {
            tx.rollback().await?;
        }
        self.lock_list.clear();
        self.cache_internal_op_list.clear();
        self.cache_op_list.clear();
        self.callback_list.clear();
        Ok(())
    }

    pub async fn begin_read_tx(&mut self) -> Result<()> {
        self.has_read_tx += 1;
        @%- for db in config.outer_db() %@
        self._@{ db }@_db.begin_read_tx().await?;
        @%- endfor %@
        Ok(())
    }

    pub fn has_read_tx(&self) -> bool {
        self.has_read_tx > 0
    }

    pub async fn get_read_tx(&mut self) -> Result<&mut Transaction<'static, DbType>> {
        ensure!(self.has_read_tx(), "No transaction is active");
        match self.read_tx.entry(self.shard_id) {
            Entry::Occupied(tx) => Ok(tx.into_mut()),
            Entry::Vacant(v) => {
                let mut tx = Self::acquire_reader_tx(self.shard_id).await?;
                set_read_tx_isolation(&mut tx).await?;
                Ok(v.insert(tx))
            }
        }
    }

    pub fn release_read_tx(&mut self) -> Result<()> {
        ensure!(self.has_read_tx(), "No read transaction is active.");
        @%- for db in config.outer_db() %@
        self._@{ db }@_db.release_read_tx()?;
        @%- endfor %@
        self.has_read_tx -= 1;
        if self.has_read_tx == 0 {
            self.read_tx.clear();
        }
        Ok(())
    }
    @%- if !config.force_disable_cache %@

    #[allow(dead_code)]
    pub(crate) async fn begin_cache_tx(&mut self) -> Result<()> {
        if self.cache_tx.is_empty() {
            let mut tx = Self::acquire_cache_tx(self.shard_id).await?;
            set_read_tx_isolation(&mut tx).await?;
            let sql = "SELECT seq FROM _sequence where id = 2";
            let sync: (u64,) = sqlx::query_as(sql).fetch_one(tx.as_mut()).await?;
            self.cache_tx.insert(self.shard_id, (sync.0, tx));
        }
        Ok(())
    }

    #[allow(dead_code)]
    pub(crate) fn has_cache_tx(&self) -> bool {
        !self.cache_tx.is_empty()
    }

    #[allow(dead_code)]
    pub(crate) async fn get_cache_tx(&mut self) -> Result<&mut Transaction<'static, DbType>> {
        ensure!(!self.cache_tx.is_empty(), "No transaction is active");
        match self.cache_tx.entry(self.shard_id) {
            Entry::Occupied(tx) => Ok(&mut tx.into_mut().1),
            Entry::Vacant(v) => {
                let mut tx = Self::acquire_cache_tx(self.shard_id).await?;
                set_read_tx_isolation(&mut tx).await?;
                let sql = "SELECT seq FROM _sequence where id = 2";
                let sync: (u64,) = sqlx::query_as(sql).fetch_one(tx.as_mut()).await?;
                Ok(&mut v.insert((sync.0, tx)).1)
            }
        }
    }

    #[allow(dead_code)]
    pub(crate) fn release_cache_tx(&mut self) {
        self.cache_tx.clear();
    }
    @%- endif %@

    pub fn reset_tx(&mut self) {
        self.save_point.clear();
        self.has_tx = false;
        self.tx.clear();
        self.lock_list.clear();
        self.cache_internal_op_list.clear();
        self.cache_op_list.clear();
        self.callback_list.clear();
        self.has_read_tx = 0;
        self.read_tx.clear();
        @%- if !config.force_disable_cache %@
        self.cache_tx.clear();
        @%- endif %@
        @%- for db in config.outer_db() %@
        self._@{ db }@_db.reset_tx();
        @%- endfor %@
    }
    @%- if config.use_sequence %@

    pub async fn sequence(&mut self, num: u64) -> Result<u64> {
        let mut cur = SEQUENCE[self.shard_id() as usize].lock().await;
        let (seq, ceiling) = *cur;
        if seq + num > ceiling {
            let base = SEQUENCE_FETCH_NUM.load(Ordering::Relaxed) as u64;
            let fetch_num = (num / base + 1) * base;
            let sql = "UPDATE _sequence SET seq = LAST_INSERT_ID(seq + ?) where id = ?;";
            let query = sqlx::query(sql).bind(fetch_num).bind(ID_OF_SEQUENCE);
            let result = query.execute(self.acquire_writer().await?.as_mut()).await?;
            let ceiling = result.last_insert_id();
            let ret = ceiling - fetch_num + 1;
            *cur = (ret + num - 1, ceiling);
            Ok(ret)
        } else {
            let ret = seq + 1;
            *cur = (seq + num, ceiling);
            Ok(ret)
        }
    }
    @%- endif %@

    pub(crate) async fn inc_all_cache_sync() -> FxHashMap<ShardId, u64> {
        let mut sync_map = FxHashMap::default();
        for shard_id in DbConn::shard_num_range() {
            sync_map.insert(shard_id, Self::inc_cache_sync(shard_id).await);
        }
        sync_map
    }
    @%- if !config.force_disable_cache %@

    async fn ___inc_cache_sync(mut pool: PoolConnection<DbType>) -> Result<u64> {
        let sql = "UPDATE _sequence SET seq = LAST_INSERT_ID(seq + ?) where id = ?;";
        let query = sqlx::query(sql).bind(1).bind(ID_OF_CACHE_SYNC);
        let result = query.execute(pool.as_mut()).await?;
        Ok(result.last_insert_id())
    }

    async fn __inc_cache_sync(shard_id: ShardId) -> Result<u64> {
        let writer = Self::_acquire_writer(shard_id).await?;
        Self::___inc_cache_sync(writer).await
    }

    async fn _inc_cache_sync(shard_id: ShardId) -> u64 {
        match Self::__inc_cache_sync(shard_id).await {
            Ok(sync) => sync,
            Err(e) => {
                log::warn!("{}", e);
                if Self::is_retryable_error(&e) {
                    Self::__inc_cache_sync(shard_id).await.unwrap_or(0)
                } else {
                    0
                }
            }
        }
    }

    pub(crate) async fn inc_cache_sync(shard_id: ShardId) -> u64 {
        static CACHE_SYNC: Lazy<Vec<Mutex<()>>> =
            Lazy::new(|| DbConn::shard_num_range().map(|_| Mutex::new(())).collect());
        static CACHE_SYNC_QUEUE: Lazy<Vec<SegQueue<Arc<AtomicU64>>>> =
            Lazy::new(|| DbConn::shard_num_range().map(|_| SegQueue::new()).collect());
        let buf = Arc::new(AtomicU64::new(u64::MAX));
        CACHE_SYNC_QUEUE[shard_id as usize].push(Arc::clone(&buf));
        let _lock = CACHE_SYNC[shard_id as usize].lock().await;
        let sync = buf.load(Ordering::SeqCst);
        if sync != u64::MAX {
            return sync;
        }
        let mut list = Vec::new();
        while let Some(b) = CACHE_SYNC_QUEUE[shard_id as usize].pop() {
            list.push(b);
        }
        let sync = Self::_inc_cache_sync(shard_id).await;
        for b in list {
            b.store(sync, Ordering::Release);
        }
        sync
    }
    @%- else %@

    pub(crate) async fn inc_cache_sync(_shard_id: ShardId) -> u64 {
        0
    }
    @%- endif %@

    pub fn max_connections_for_write() -> u32 {
        MAX_CONNECTIONS_FOR_WRITE.load(Ordering::Relaxed)
    }
    pub fn max_connections_for_read() -> u32 {
        MAX_CONNECTIONS_FOR_READ.load(Ordering::Relaxed)
    }
    pub fn max_connections_for_cache() -> u32 {
        MAX_CONNECTIONS_FOR_CACHE.load(Ordering::Relaxed)
    }

    pub(crate) fn _has_update_notice() -> bool {
        NOTIFY_RECEIVER_COUNT.load(Ordering::Relaxed) > 0
    }
    pub(crate) async fn _push_update_notice(
        table: TableName,
        op: NotifyOp,
        id: &impl serde::Serialize,
    ) {
        let id = serde_json::to_string(id).unwrap();
        let mut notify_list = NOTIFY_LIST.write().await;
        let mut list = notify_list.take().unwrap_or_default();
        let key = (table, id);
        if let Some(_op) = list.get(&key) {
            if op < *_op {
                list.insert(key, op);
            }
        } else {
            list.insert(key, op);
        }
        notify_list.replace(list);
    }
    pub(crate) async fn _publish_update_notice() {
        let list = NOTIFY_LIST.write().await.take();
        if let Some(list) = list {
            for f in NOTIFY_RECEIVER.read().await.iter() {
                for (row, op) in &list {
                    f(row.0, *op, &row.1);
                }
            }
        }
    }
    @%- if config.use_update_notice %@
    pub async fn subscribe_update_notice(f: NotifyFn) {
        let mut receivers = NOTIFY_RECEIVER.write().await;
        receivers.push(f);
        NOTIFY_RECEIVER_COUNT.store(receivers.len(), Ordering::SeqCst);
    }
    @%- endif %@

    /// Obtain a lock during a transaction
    /// Locks are released along with transaction commits or rollbacks.
    /// A negative timeout value means infinite timeout.
    pub async fn lock(&mut self, key: &str, timeout_secs: i32) -> Result<()> {
        static LOCK_MAP: Mutex<BTreeMap<String, Arc<Semaphore>>> = Mutex::const_new(BTreeMap::new());
        let lock = LOCK_MAP
            .lock()
            .await
            .entry(key.to_string())
            .or_insert_with(|| Arc::new(Semaphore::new(1)))
            .clone()
            .acquire_owned();
        let (semaphore, timeout) = if timeout_secs >= 0 {
            let start = std::time::Instant::now();
            let semaphore =
                tokio::time::timeout(Duration::from_secs(timeout_secs as u64), lock).await??;
            (semaphore, timeout_secs - start.elapsed().as_secs() as i32)
        } else {
            (lock.await?, timeout_secs)
        };
        let mut conn = self.acquire_writer().await?;
        let result: (Option<i64>,) = sqlx::query_as("SELECT GET_LOCK(?, ?)")
            .bind(key)
            .bind(timeout)
            .fetch_one(conn.as_mut())
            .await?;
        if result.0 == Some(1) {
            self.lock_list.push(DbLock {
                conn: Some(conn),
                _semaphore: semaphore,
            });
            Ok(())
        } else {
            Err(senax_common::err::LockFailed::new(key.to_string()).into())
        }
    }

    pub fn take_lock(&mut self) -> Option<DbLock> {
        self.lock_list.pop()
    }

    pub fn is_retryable_error(err: &anyhow::Error) -> bool {
        if let Some(err) = err.downcast_ref::<sqlx::Error>() {
            match err {
                sqlx::Error::Io(..) => {
                    log::warn!("{}", err);
                    return true;
                }
                sqlx::Error::WorkerCrashed => {
                    log::warn!("{}", err);
                    return true;
                }
                _ => {}
            }
        }
        false
    }
}

pub struct DbLock {
    conn: Option<PoolConnection<DbType>>,
    _semaphore: tokio::sync::OwnedSemaphorePermit,
}
impl Drop for DbLock {
    fn drop(&mut self) {
        let mut conn = self.conn.take().unwrap();
        tokio::spawn(async move {
            if let Err(e) = sqlx::query("DO RELEASE_ALL_LOCKS()")
                .fetch_all(conn.as_mut())
                .await
            {
                log::error!("RELEASE_LOCK ERROR: {}", e);
            }
        });
    }
}

impl Drop for DbConn {
    fn drop(&mut self) {
        if cfg!(debug_assertions)
            && (self.has_tx() || !self.callback_list.is_empty() || !self.cache_op_list.is_empty())
        {
            log::debug!("implicit rollback");
        }
    }
}

#[cfg(feature = "use_google_resolver")]
async fn resolve_ips(host: &str, port: u16) -> std::io::Result<Vec<SocketAddr>> {
    use hickory_resolver::config::{ResolverConfig, ResolverOpts};
    use hickory_resolver::TokioAsyncResolver;
    let resolver = TokioAsyncResolver::tokio(ResolverConfig::google(), ResolverOpts::default());
    let response = resolver.lookup_ip(host).await?;
    Ok(response
        .iter()
        .map(|ip| SocketAddr::new(ip, port))
        .collect())
}

#[allow(clippy::explicit_counter_loop)]
async fn lookup(url: &url::Url) -> Result<Vec<SocketAddr>> {
    use std::net::{IpAddr, Ipv4Addr};
    let port = url.port().unwrap_or(0);
    if let Some(host) = url.host() {
        match host {
            url::Host::Domain(domain) => {
                let mut set = FxHashSet::default();
                let mut _count = 0;
                for v in tokio::net::lookup_host(format!("{}:{}", domain, port)).await? {
                    _count += 1;
                    set.insert(v);
                }
                #[cfg(feature = "use_google_resolver")]
                if domain.contains('.') && _count != set.len() {
                    let mut task = JoinSet::new();
                    for _ in 0..100 {
                        let domain = domain.to_string();
                        task.spawn(async move { resolve_ips(&domain, port).await });
                    }
                    while let Some(r) = task.join_next().await {
                        if let Ok(Ok(r)) = r {
                            for v in r {
                                set.insert(v);
                                if _count == set.len() {
                                    break;
                                }
                            }
                        }
                    }
                }
                if set.is_empty() {
                    anyhow::bail!("Host not found: {}", url);
                }
                Ok(set
                    .into_iter()
                    .filter(|v| {
                        if v.is_ipv4() {
                            !USE_IPV6_ONLY.load(Ordering::Relaxed)
                        } else {
                            !USE_IPV4_ONLY.load(Ordering::Relaxed)
                        }
                    })
                    .collect())
            }
            url::Host::Ipv4(ip) => Ok(vec![SocketAddr::new(IpAddr::V4(ip), port)]),
            url::Host::Ipv6(ip) => Ok(vec![SocketAddr::new(IpAddr::V6(ip), port)]),
        }
    } else {
        let ip = Ipv4Addr::new(127, 0, 0, 1);
        Ok(vec![SocketAddr::new(IpAddr::V4(ip), port)])
    }
}

async fn check_if_writable(options: &DbConnectOptions) -> Result<Option<bool>> {
    use sqlx::{Acquire, Connection};
    let mut conn = match DbConnection::connect_with(options).await {
        Err(sqlx::Error::Database(e)) => {
            // Assumes too many connections
            warn!("{}", e);
            return Ok(None);
        }
        Err(e) => {
            anyhow::bail!(e);
        }
        Ok(conn) => conn,
    };
    let conn = conn.acquire().await?;
    let row: (i8,) = sqlx::query_as(CHECK_SQL).fetch_one(conn).await?;
    Ok(Some(row.0 == CHECK_SQL_WRITABLE_RESULT))
}

#[allow(clippy::too_many_arguments)]
async fn check_connection(
    urls: &str,
    user: &Option<String>,
    pw: &Option<String>,
    shard_id: usize,
    removals: Option<Arc<Mutex<AddrPoolMap>>>,
    pool_option: Option<sqlx::pool::PoolOptions<connection::DbType>>,
    pool_collection: Option<&'static OnceCell<Vec<Arc<RwLock<AddrPoolMap>>>>>,
    target: Target,
) -> Result<Vec<(SocketAddr, DbConnectOptions, Option<bool>)>> {
    let re = regex::Regex::new(r"[ \t]+").unwrap();
    let mut join_set: JoinSet<Result<Vec<(SocketAddr, DbConnectOptions)>>> = JoinSet::new();
    for url in re.split(urls) {
        let mut url = url::Url::parse(url)?;
        let user = user.clone();
        let pw = pw.clone();
        join_set.spawn(async move {
            let mut list = Vec::new();
            for addr in lookup(&url).await? {
                let host = if addr.is_ipv4() {
                    addr.ip().to_string()
                } else {
                    format!("[{}]", addr.ip())
                };
                url.set_host(Some(&host))?;
                let options: DbConnectOptions = url.as_str().parse()?;
                let options = if let Some(ref user) = user {
                    options.username(user)
                } else {
                    options
                };
                let options = if let Some(ref pw) = pw {
                    options.password(pw)
                } else {
                    options
                };
                list.push((addr, crate::db_options(options)));
            }
            Ok(list)
        });
    }
    let mut list = Vec::new();
    let mut check_set = FxHashSet::default();
    while let Some(r) = join_set.join_next().await {
        match r? {
            Ok(r) => {
                for (addr, options) in r {
                    if check_set.insert(addr) {
                        list.push((addr, options));
                    }
                }
            }
            Err(e) => {
                warn!("{}", e);
            }
        }
    }

    let mut join_set = JoinSet::new();
    for (addr, options) in list {
        join_set.spawn(async move {
            match tokio::time::timeout(
                Duration::from_secs(crate::CHECK_CONNECTION_TIMEOUT),
                check_if_writable(&options),
            )
            .await
            {
                Ok(res) => match res {
                    Ok(writable) => {
                        let options = options.log_statements(LevelFilter::Debug);
                        if target == Target::Write && writable == Some(false) {
                            None
                        } else {
                            Some((addr, options, writable))
                        }
                    }
                    Err(e) => {
                        warn!("{}: {}", e, addr);
                        None
                    }
                },
                Err(_) => {
                    warn!(
                        "error communicating with database: Connection timeout: {}",
                        addr
                    );
                    None
                }
            }
        });
    }
    if target == Target::Write {
        while let Some(r) = join_set.join_next().await {
            if let Some(r) = r? {
                return Ok(vec![r]);
            }
        }
        Ok(vec![])
    } else {
        let (tx, mut rx) = mpsc::channel(10);
        let removals = removals.unwrap();
        let pool_option = pool_option.unwrap();
        tokio::spawn(async move {
            while let Some(r) = join_set.join_next().await {
                if let Ok(Some(r)) = r {
                    if let Err(r) = tx.send(r).await {
                        if let Some(pool_collection) = pool_collection {
                            let (addr, options, writable) = r.0;
                            if writable == Some(true) {
                                continue;
                            }
                            if let Some(old) = removals.lock().await.remove(&addr) {
                                let mut pools =
                                    pool_collection.get().unwrap()[shard_id].write().await;
                                if writable == Some(false) {
                                    pools.retain(|_, (_, w)| *w != Some(true));
                                }
                                pools.insert(addr, (old.0, writable));
                                continue;
                            }
                            if writable.is_some() {
                                let pool = reader_connect_with(pool_option.clone(), options).await;
                                let pool = match pool {
                                    Ok(pool) => pool,
                                    Err(e) => {
                                        warn!("{}", e);
                                        continue;
                                    }
                                };
                                log::info!("add connection for {}: {}", target, addr);
                                let mut pools =
                                    pool_collection.get().unwrap()[shard_id].write().await;
                                if writable == Some(false) {
                                    pools.retain(|_, (_, w)| *w != Some(true));
                                }
                                let pool = Arc::new(PoolInfo {
                                    addr,
                                    target,
                                    inner: pool,
                                });
                                pools.insert(addr, (pool, writable));
                            }
                        }
                    }
                }
            }
        });
        let mut result = Vec::new();
        let mut timeout = false;
        // At least one connection is required, so tokio::time::timeout cannot be used.
        let sleep = tokio::time::sleep(Duration::from_millis(500));
        tokio::pin!(sleep);
        loop {
            tokio::select!(
                r = rx.recv() => {
                    if let Some(r) = r {
                        result.push(r);
                        if timeout {
                            break;
                        }
                    } else {
                        break;
                    }
                },
                _ = &mut sleep => {
                    if result.is_empty() {
                        timeout = true;
                    } else {
                        break;
                    }
                }
            )
        }
        Ok(result)
    }
}

#[allow(clippy::too_many_arguments)]
async fn reset_writer_pool(
    s_url: &str,
    s_user: &Option<String>,
    s_pw: &Option<String>,
    r_url: &str,
    r_user: &Option<String>,
    r_pw: &Option<String>,
    pool_option: sqlx::pool::PoolOptions<connection::DbType>,
    pool_collection: &'static RwLock<Vec<Option<(SocketAddr, DbPool)>>>,
    shard_id: Option<ShardId>,
) -> Result<()> {
    let r_url_list: Vec<_> = split_shard!(r_url).collect();
    let mut join_set: JoinSet<Result<()>> = JoinSet::new();
    for (idx, s_url) in split_shard!(s_url).enumerate() {
        if let Some(shard_id) = shard_id {
            if shard_id as usize != idx {
                continue;
            }
        }
        let r_url = r_url_list
            .get(idx)
            .context("The number of shards differs between SOURCE and REPLICA.")?;
        if idx >= DbConn::shard_num() {
            break;
        }
        let _s_url = s_url.to_owned();
        let _s_user = s_user.to_owned();
        let _s_pw = s_pw.to_owned();
        let _r_url = (*r_url).to_owned();
        let _r_user = r_user.to_owned();
        let _r_pw = r_pw.to_owned();
        let pool_option = pool_option.clone();
        let check_fast_failover = ENABLE_FAST_FAILOVER.load(Ordering::Relaxed) && s_url != *r_url;
        join_set.spawn(async move {
            let mut join = JoinSet::new();
            join.spawn(async move {
                check_connection(
                    &_s_url,
                    &_s_user,
                    &_s_pw,
                    idx,
                    None,
                    None,
                    None,
                    Target::Write,
                )
                .await
            });
            let (tx, mut rx) = tokio::sync::mpsc::channel(2);
            if check_fast_failover {
                join.spawn(async move {
                    let _ = tokio::time::timeout(Duration::from_millis(500), rx.recv()).await;
                    check_connection(
                        &_r_url,
                        &_r_user,
                        &_r_pw,
                        idx,
                        None,
                        None,
                        None,
                        Target::Write,
                    )
                    .await
                });
            }
            let mut c = Vec::new();
            while let Some(r) = join.join_next().await {
                match r? {
                    Ok(r) => {
                        if r.is_empty() {
                            let _ = tx.send(0).await;
                        } else {
                            c = r;
                            break;
                        }
                    }
                    Err(e) => {
                        warn!("{}", e);
                    }
                }
            }

            if let Some((addr, options, writable)) = c.pop() {
                if let Some(old) = &pool_collection.read().await[idx] {
                    if old.0 == addr {
                        return Ok(());
                    }
                }
                if writable == Some(true) {
                    let new_pool = writer_connect_with(pool_option, options).await?;
                    let pool = &mut pool_collection.write().await[idx];
                    @%- if !config.force_disable_cache %@
                    #[cfg(not(feature = "cache_update_only"))]
                    if pool.is_some() {
                        let new_pool = new_pool.clone();
                        tokio::spawn(async move {
                            if let Ok(pool) = new_pool.acquire().await {
                                let sync = DbConn::___inc_cache_sync(pool).await.unwrap_or(0);
                                @%- for (name, defs) in groups %@
                                models::@{ name|snake|to_var_name }@::_clear_cache(idx as ShardId, sync, false).await;
                                @%- endfor %@
                            }
                        });
                    }
                    @%- endif %@
                    log::info!("switch connection for {}: {}", Target::Write, addr);
                    *pool = Some((addr, new_pool));
                }
            }
            Ok(())
        });
    }
    while let Some(r) = join_set.join_next().await {
        if let Err(e) = r? {
            log::error!("{}", e);
        }
    }
    Ok(())
}

async fn writer_connect_with(
    pool_option: sqlx::pool::PoolOptions<connection::DbType>,
    options: DbConnectOptions,
) -> Result<DbPool> {
    Ok(pool_option
        .after_connect(|conn, _meta| {
            Box::pin(async move {
                let sql = if let Some(iso) = TX_ISOLATION {
                    format!(
                        "{};SET SESSION sql_mode=(SELECT CONCAT(@@sql_mode,',ANSI_QUOTES,PIPES_AS_CONCAT,NO_ENGINE_SUBSTITUTION')),time_zone='+00:00',NAMES utf8mb4 COLLATE utf8mb4_unicode_ci;SET SESSION TRANSACTION ISOLATION LEVEL {};",
                        CHECK_SQL,
                        iso,
                    )
                } else {
                    format!(
                        "{};SET SESSION sql_mode=(SELECT CONCAT(@@sql_mode,',ANSI_QUOTES,PIPES_AS_CONCAT,NO_ENGINE_SUBSTITUTION')),time_zone='+00:00',NAMES utf8mb4 COLLATE utf8mb4_unicode_ci;",
                        CHECK_SQL,
                    )
                };
                let mut res = conn.fetch_many(&*sql);
                use ::futures::TryStreamExt;
                let mut first = true;
                let mut writable = false;
                while let Some(res) = res.try_next().await? {
                    if first {
                        first = false;
                        if let sqlx::Either::Right(res) = res {
                            let i:i8 = res.try_get(0)?;
                            writable = i == CHECK_SQL_WRITABLE_RESULT;
                        }
                    }
                }
                if !writable {
                    return Err(sqlx::Error::Protocol("There are no writable database connections.(code:4)".to_string()));
                }
                Ok(())
            })
        })
        .connect_with(options)
        .await?)
}

#[allow(clippy::too_many_arguments)]
async fn reset_reader_pool(
    r_url: &str,
    r_user: &Option<String>,
    r_pw: &Option<String>,
    s_url: &str,
    s_user: &Option<String>,
    s_pw: &Option<String>,
    pool_option: sqlx::pool::PoolOptions<connection::DbType>,
    pool_collection: &'static OnceCell<Vec<Arc<RwLock<AddrPoolMap>>>>,
    shard_id: Option<ShardId>,
    target: Target,
) -> Result<()> {
    let s_url_list: Vec<_> = split_shard!(s_url).collect();
    let mut join_set: JoinSet<Result<()>> = JoinSet::new();
    for (idx, r_url) in split_shard!(r_url).enumerate() {
        if let Some(shard_id) = shard_id {
            if shard_id as usize != idx {
                continue;
            }
        }
        let s_url = s_url_list
            .get(idx)
            .context("The number of shards differs between SOURCE and REPLICA.")?;
        if idx >= DbConn::shard_num() {
            break;
        }
        let _s_url = (*s_url).to_owned();
        let _s_user = s_user.to_owned();
        let _s_pw = s_pw.to_owned();
        let _r_url = r_url.to_owned();
        let _r_user = r_user.to_owned();
        let _r_pw = r_pw.to_owned();
        let pool_option = pool_option.clone();
        let check_fast_failover = ENABLE_FAST_FAILOVER.load(Ordering::Relaxed) && r_url != *s_url;
        join_set.spawn(async move {
            let removals = Arc::new(Mutex::new(AddrPoolMap::default()));
            let mut removals_lock = removals.lock().await;
            let removals_ = removals.clone();
            let option1 = pool_option.clone();
            let option2 = option1.clone();
            let mut join = JoinSet::new();
            join.spawn(async move {
                check_connection(
                    &_r_url,
                    &_r_user,
                    &_r_pw,
                    idx,
                    Some(removals_),
                    Some(option1),
                    Some(pool_collection),
                    target,
                )
                .await
            });
            let (tx, mut rx) = tokio::sync::mpsc::channel(2);
            if check_fast_failover {
                let removals_ = removals.clone();
                join.spawn(async move {
                    let _ = tokio::time::timeout(Duration::from_millis(500), rx.recv()).await;
                    check_connection(
                        &_s_url,
                        &_s_user,
                        &_s_pw,
                        idx,
                        Some(removals_),
                        Some(option2),
                        Some(pool_collection),
                        target,
                    )
                    .await
                });
            }
            let mut c = Vec::new();
            while let Some(r) = join.join_next().await {
                match r? {
                    Ok(r) => {
                        if r.is_empty() {
                            let _ = tx.send(0).await;
                        } else {
                            c = r;
                            break;
                        }
                    }
                    Err(e) => {
                        warn!("{}", e);
                    }
                }
            }
            if !c.is_empty() {
                let new_addrs: FxHashSet<_> = c.iter().map(|v| v.0).collect();
                let old_addrs: FxHashSet<_> = {
                    let pools = pool_collection.get().unwrap()[idx].read().await;
                    pools.keys().cloned().collect()
                };
                for (addr, options, writable) in c {
                    if old_addrs.contains(&addr) {
                        let mut pools = pool_collection.get().unwrap()[idx].write().await;
                        pools.entry(addr).and_modify(|(_, w)| *w = writable);
                    } else if writable.is_some() {
                        match reader_connect_with(pool_option.clone(), options).await {
                            Ok(pool) => {
                                log::info!("add connection for {}: {}", target, addr);
                                let mut pools = pool_collection.get().unwrap()[idx].write().await;
                                let pool = Arc::new(PoolInfo {
                                    addr,
                                    target,
                                    inner: pool,
                                });
                                pools.insert(addr, (pool, writable));
                            }
                            Err(e) => {
                                warn!("{}", e);
                            }
                        }
                    }
                }
                let mut pools = pool_collection.get().unwrap()[idx].write().await;
                removals_lock.clone_from(&pools);
                removals_lock.retain(|k, _| !new_addrs.contains(k));
                pools.retain(|k, _| new_addrs.contains(k));
                let has_readable = pools.iter().any(|(_, v)| v.1 == Some(false));
                if has_readable {
                    // remove writable connections
                    pools.retain(|_, v| v.1 != Some(true));
                }
            }
            Ok(())
        });
    }
    while let Some(r) = join_set.join_next().await {
        if let Err(e) = r? {
            log::error!("{}", e);
        }
    }
    Ok(())
}

async fn reader_connect_with(
    pool_option: sqlx::pool::PoolOptions<connection::DbType>,
    options: DbConnectOptions,
) -> Result<DbPool> {
    Ok(pool_option
        .after_connect(|conn, _meta| {
            Box::pin(async move {
                if let Some(iso) = READ_TX_ISOLATION {
                    conn.execute(&*format!(
                        "SET SESSION sql_mode=(SELECT CONCAT(@@sql_mode,',ANSI_QUOTES,PIPES_AS_CONCAT,NO_ENGINE_SUBSTITUTION')),time_zone='+00:00',NAMES utf8mb4 COLLATE utf8mb4_unicode_ci;SET SESSION TRANSACTION ISOLATION LEVEL {}, READ ONLY;",
                        iso
                    ))
                    .await?;
                } else {
                    conn.execute("SET SESSION sql_mode=(SELECT CONCAT(@@sql_mode,',ANSI_QUOTES,PIPES_AS_CONCAT,NO_ENGINE_SUBSTITUTION')),time_zone='+00:00',NAMES utf8mb4 COLLATE utf8mb4_unicode_ci;SET SESSION TRANSACTION READ ONLY;").await?;
                }
                Ok(())
            })
        })
        .connect_with(options)
        .await?)
}

#[allow(dead_code)]
pub(crate) fn _is_retryable_error<T>(result: Result<T>, table: &str) -> bool {
    if let Err(err) = result {
        if let Some(err) = err.downcast_ref::<sqlx::Error>() {
            match err {
                sqlx::Error::Io(..) => {
                    // retry all
                    log::error!(table = table; "{}", err);
                    return true;
                }
                sqlx::Error::WorkerCrashed => {
                    // retry all
                    log::error!(table = table; "{}", err);
                    return true;
                }
                _ => {
                    log::error!(table = table; "{}", err);
                }
            }
        } else {
            log::error!(table = table; "{}", err);
        }
    }
    false
}

fn env_u32(etcd: &FxHashMap<String, String>, name: &str, default: &str) -> Result<u32> {
    let v = etcd.get(name).cloned();
    let v = v.unwrap_or_else(|| env::var(name).unwrap_or_else(|_| default.to_owned()));
    v.parse().with_context(|| format!("{} parse error", name))
}
fn env_opt_str(etcd: &FxHashMap<String, String>, name: &str) -> Option<String> {
    let v = etcd.get(name).cloned();
    v.or_else(|| env::var(name).ok())
}
fn env_urls(etcd: &FxHashMap<String, String>, name: &str) -> Result<Option<String>> {
    let mut urls = etcd.get(name).cloned();
    if urls.is_none() && !etcd.is_empty() {
        let prefix = format!("{}/", name);
        let mut values = BTreeMap::new();
        for (k, v) in etcd {
            if k.starts_with(&prefix) {
                let i: usize = k.trim_start_matches(&prefix).parse()?;
                values.insert(i, v);
            }
        }
        if !values.is_empty() {
            let vec: Vec<_> = values.values().map(|v| v.to_string()).collect();
            urls = Some(vec.join(" "));
        }
    }
    Ok(urls.or_else(|| env::var(name).ok()))
}

async fn set_tx_isolation(_tx: &mut Transaction<'_, DbType>) -> Result<()> {
    // postgresqlはBEGINのあとにSET TRANSACTION
    Ok(())
}

async fn set_read_tx_isolation(_tx: &mut Transaction<'_, DbType>) -> Result<()> {
    Ok(())
}
@{-"\n"}@