// Open starts the 2PC MM service. If the metadata database or tables
// are not present, they are created.
func (tpc *TwoPC) Open(sidecarDBName string, dbaparams *sqldb.ConnParams) {
	conn, err := dbconnpool.NewDBConnection(dbaparams, stats.NewTimings(""))
	if err != nil {
		panic(err)
	}
	defer conn.Close()
	statements := []string{
		sqlTurnoffBinlog,
		fmt.Sprintf(sqlCreateSidecarDB, sidecarDBName),
		fmt.Sprintf(sqlCreateTableRedoLogTransaction, sidecarDBName),
		fmt.Sprintf(sqlCreateTableRedoLogStatement, sidecarDBName),
		fmt.Sprintf(sqlCreateTableTransaction, sidecarDBName),
		fmt.Sprintf(sqlCreateTableParticipant, sidecarDBName),
	}
	for _, s := range statements {
		if _, err := conn.ExecuteFetch(s, 0, false); err != nil {
			panic(NewTabletError(vtrpcpb.ErrorCode_INTERNAL_ERROR, err.Error()))
		}
	}
	tpc.insertRedoTx = buildParsedQuery(
		"insert into `%s`.redo_log_transaction(dtid, state, time_created) values (%a, 'Prepared', %a)",
		sidecarDBName, ":dtid", ":time_created")
	tpc.insertRedoStmt = buildParsedQuery(
		"insert into `%s`.redo_log_statement(dtid, id, statement) values %a",
		sidecarDBName, ":vals")
	tpc.deleteRedoTx = buildParsedQuery(
		"delete from `%s`.redo_log_transaction where dtid = %a",
		sidecarDBName, ":dtid")
	tpc.deleteRedoStmt = buildParsedQuery(
		"delete from `%s`.redo_log_statement where dtid = %a",
		sidecarDBName, ":dtid")
	tpc.readPrepared = fmt.Sprintf(
		"select s.dtid, s.id, s.statement from `%s`.redo_log_transaction t "+
			"join `%s`.redo_log_statement s on t.dtid = s.dtid "+
			"where t.state = 'Prepared' order by s.dtid, s.id",
		sidecarDBName, sidecarDBName)

	tpc.insertTransaction = buildParsedQuery(
		"insert into `%s`.transaction(dtid, state, time_created, time_updated) values (%a, 'Prepare', %a, %a)",
		sidecarDBName, ":dtid", ":cur_time", ":cur_time")
	tpc.insertParticipants = buildParsedQuery(
		"insert into `%s`.participant(dtid, id, keyspace, shard) values %a",
		sidecarDBName, ":vals")
	tpc.transition = buildParsedQuery(
		"update `%s`.transaction set state = %a where dtid = %a and state = 'Prepare'",
		sidecarDBName, ":state", ":dtid")
	tpc.deleteTransaction = buildParsedQuery(
		"delete from `%s`.transaction where dtid = %a",
		sidecarDBName, ":dtid")
	tpc.deleteParticipants = buildParsedQuery(
		"delete from `%s`.participant where dtid = %a",
		sidecarDBName, ":dtid")
	tpc.readTransaction = buildParsedQuery(
		"select dtid, state, time_created, time_updated from `%s`.transaction where dtid = %a",
		sidecarDBName, ":dtid")
	tpc.readParticipants = buildParsedQuery(
		"select keyspace, shard from `%s`.participant where dtid = %a",
		sidecarDBName, ":dtid")
}

func (dbc *DBConn) reconnect() error {
	dbc.conn.Close()
	newConn, err := dbconnpool.NewDBConnection(dbc.info, dbc.queryServiceStats.MySQLStats)
	if err != nil {
		return err
	}
	dbc.conn = newConn
	return nil
}

// CheckMySQL returns true if we can connect to MySQL.
func (qe *QueryEngine) CheckMySQL() bool {
	conn, err := dbconnpool.NewDBConnection(&qe.dbconfigs.App.ConnParams, qe.queryServiceStats.MySQLStats)
	if err != nil {
		if IsConnErr(err) {
			return false
		}
		log.Warningf("checking MySQL, unexpected error: %v", err)
		return true
	}
	conn.Close()
	return true
}

// allowQueries starts the query service.
// If the state is other than StateServing or StateNotServing, it fails.
// If allowQuery succeeds, the resulting state is StateServing.
// Otherwise, it reverts back to StateNotServing.
// While allowQuery is running, the state is set to StateInitializing.
// If waitForMysql is set to true, allowQueries will not return
// until it's able to connect to mysql.
// No other operations are allowed when allowQueries is running.
func (sq *SqlQuery) allowQueries(target *pb.Target, dbconfigs *dbconfigs.DBConfigs, schemaOverrides []SchemaOverride, mysqld mysqlctl.MysqlDaemon) (err error) {
	sq.mu.Lock()
	if sq.state == StateServing {
		sq.mu.Unlock()
		return nil
	}
	if sq.state != StateNotServing {
		state := sq.state
		sq.mu.Unlock()
		return NewTabletError(ErrFatal, vtrpc.ErrorCode_INTERNAL_ERROR, "cannot start query service, current state: %s", state)
	}
	// state is StateNotServing
	sq.setState(StateInitializing)
	sq.mu.Unlock()

	c, err := dbconnpool.NewDBConnection(&dbconfigs.App.ConnParams, sq.qe.queryServiceStats.MySQLStats)
	if err != nil {
		log.Infof("allowQueries failed: %v", err)
		sq.mu.Lock()
		sq.setState(StateNotServing)
		sq.mu.Unlock()
		return err
	}
	c.Close()

	defer func() {
		state := int64(StateServing)
		if x := recover(); x != nil {
			err = x.(*TabletError)
			log.Errorf("Could not start query service: %v", err)
			sq.qe.Close()
			state = StateNotServing
		}
		sq.mu.Lock()
		sq.setState(state)
		sq.mu.Unlock()
	}()

	sq.qe.Open(dbconfigs, schemaOverrides)
	// Start the invalidator after qe.
	if needInvalidator(target, dbconfigs) {
		sq.invalidator.Open(dbconfigs.App.DbName, mysqld)
	}
	sq.dbconfig = &dbconfigs.App
	sq.target = target
	sq.sessionID = Rand()
	log.Infof("Session id: %d", sq.sessionID)
	return nil
}

// NewDBConn creates a new DBConn. It triggers a CheckMySQL if creation fails.
func NewDBConn(
	cp *ConnPool,
	appParams,
	dbaParams *sqldb.ConnParams,
	qStats *QueryServiceStats) (*DBConn, error) {
	c, err := dbconnpool.NewDBConnection(appParams, qStats.MySQLStats)
	if err != nil {
		cp.checker.CheckMySQL()
		return nil, err
	}
	return &DBConn{
		conn:              c,
		info:              appParams,
		pool:              cp,
		queryServiceStats: qStats,
	}, nil
}

// allowQueries starts the query service.
// If the state is anything other than NOT_SERVING, it fails.
// If allowQuery succeeds, the resulting state is SERVING.
// Otherwise, it reverts back to NOT_SERVING.
// While allowQuery is running, the state is set to INITIALIZING.
// If waitForMysql is set to true, allowQueries will not return
// until it's able to connect to mysql.
// No other operations are allowed when allowQueries is running.
func (sq *SqlQuery) allowQueries(dbconfig *dbconfigs.DBConfig, schemaOverrides []SchemaOverride, qrs *QueryRules, mysqld *mysqlctl.Mysqld, waitForMysql bool) (err error) {
	sq.mu.Lock()
	defer sq.mu.Unlock()
	if sq.state.Get() != NOT_SERVING {
		terr := NewTabletError(FATAL, "cannot start query service, current state: %s", sq.GetState())
		return terr
	}
	// state is NOT_SERVING
	sq.setState(INITIALIZING)

	if waitForMysql {
		waitTime := time.Second
		for {
			c, err := dbconnpool.NewDBConnection(&dbconfig.ConnectionParams, mysqlStats)
			if err == nil {
				c.Close()
				break
			}
			log.Warningf("mysql.Connect() error, retrying in %v: %v", waitTime, err)
			time.Sleep(waitTime)
			// Cap at 32 seconds
			if waitTime < 30*time.Second {
				waitTime = waitTime * 2
			}
		}
	}

	defer func() {
		if x := recover(); x != nil {
			err = x.(*TabletError)
			log.Errorf("Could not start query service: %v", err)
			sq.qe.Close()
			sq.setState(NOT_SERVING)
			return
		}
		sq.setState(SERVING)
	}()

	sq.qe.Open(dbconfig, schemaOverrides, qrs, mysqld)
	sq.dbconfig = dbconfig
	sq.mysqld = mysqld
	sq.sessionId = Rand()
	log.Infof("Session id: %d", sq.sessionId)
	return nil
}

func (tsv *TabletServer) fullStart() (err error) {
	defer func() {
		if x := recover(); x != nil {
			log.Errorf("Could not start tabletserver: %v", x)
			tsv.qe.Close()
			tsv.transition(StateNotConnected)
			err = x.(error)
		}
	}()

	c, err := dbconnpool.NewDBConnection(&tsv.dbconfigs.App.ConnParams, tsv.qe.queryServiceStats.MySQLStats)
	if err != nil {
		panic(err)
	}
	c.Close()

	tsv.qe.Open(tsv.dbconfigs)
	return tsv.serveNewType()
}

func (sq *SqlQuery) fullStart() (err error) {
	defer func() {
		if x := recover(); x != nil {
			log.Errorf("Could not start tabletserver: %v", x)
			sq.qe.Close()
			sq.mu.Lock()
			sq.setState(StateNotConnected)
			sq.mu.Unlock()
			err = x.(error)
		}
	}()

	c, err := dbconnpool.NewDBConnection(&sq.dbconfigs.App.ConnParams, sq.qe.queryServiceStats.MySQLStats)
	if err != nil {
		panic(err)
	}
	c.Close()

	sq.qe.Open(sq.dbconfigs, sq.schemaOverrides)
	return sq.serveNewType()
}

// GetDbaConnection creates a new DBConnection.
func (mysqld *Mysqld) GetDbaConnection() (*dbconnpool.DBConnection, error) {
	return dbconnpool.NewDBConnection(mysqld.dba, mysqld.dbaMysqlStats)
}

// GetAllPrivsConnection creates a new DBConnection.
func (mysqld *Mysqld) GetAllPrivsConnection() (*dbconnpool.DBConnection, error) {
	return dbconnpool.NewDBConnection(mysqld.allprivs, mysqld.allprivsMysqlStats)
}

// GetAllPrivsConnection is part of the MysqlDaemon interface.
func (fmd *FakeMysqlDaemon) GetAllPrivsConnection() (*dbconnpool.DBConnection, error) {
	return dbconnpool.NewDBConnection(&sqldb.ConnParams{Engine: fmd.db.Name}, stats.NewTimings(""))
}

// GetDbaConnection is part of the MysqlDaemon interface.
func (fmd *FakeMysqlDaemon) GetDbaConnection() (*dbconnpool.DBConnection, error) {
	return dbconnpool.NewDBConnection(&sqldb.ConnParams{}, stats.NewTimings(""))
}