// CreateSession creates a new session environment.
func CreateSession(store kv.Storage) (Session, error) {
	s := &session{
		values:      make(map[fmt.Stringer]interface{}),
		store:       store,
		sid:         atomic.AddInt64(&sessionID, 1),
		debugInfos:  make(map[string]interface{}),
		maxRetryCnt: 10,
		parser:      parser.New(),
	}
	domain, err := domap.Get(store)
	if err != nil {
		return nil, errors.Trace(err)
	}
	sessionctx.BindDomain(s, domain)

	variable.BindSessionVars(s)
	variable.GetSessionVars(s).SetStatusFlag(mysql.ServerStatusAutocommit, true)

	// session implements variable.GlobalVarAccessor. Bind it to ctx.
	variable.BindGlobalVarAccessor(s, s)

	// session implements autocommit.Checker. Bind it to ctx
	autocommit.BindAutocommitChecker(s, s)
	sessionMu.Lock()
	defer sessionMu.Unlock()

	ver := getStoreBootstrapVersion(store)
	if ver == notBootstrapped {
		// if no bootstrap and storage is remote, we must use a little lease time to
		// bootstrap quickly, after bootstrapped, we will reset the lease time.
		// TODO: Using a bootstap tool for doing this may be better later.
		if !localstore.IsLocalStore(store) {
			sessionctx.GetDomain(s).SetLease(chooseMinLease(100*time.Millisecond, schemaLease))
		}

		s.SetValue(context.Initing, true)
		bootstrap(s)
		s.ClearValue(context.Initing)

		if !localstore.IsLocalStore(store) {
			sessionctx.GetDomain(s).SetLease(schemaLease)
		}

		finishBootstrap(store)
	} else if ver < currentBootstrapVersion {
		s.SetValue(context.Initing, true)
		upgrade(s)
		s.ClearValue(context.Initing)
	}

	// TODO: Add auth here
	privChecker := &privileges.UserPrivileges{}
	privilege.BindPrivilegeChecker(s, privChecker)
	return s, nil
}

func (dm *domainMap) Get(store kv.Storage) (d *domain.Domain, err error) {
	key := store.UUID()
	dm.mu.Lock()
	defer dm.mu.Unlock()
	d = dm.domains[key]
	if d != nil {
		return
	}

	if SchemaLease <= minSchemaLease {
		SchemaLease = minSchemaLease
	}

	lease := SchemaLease
	// if storage is local storage, we may in test environment or
	// run server in single machine mode, so we don't need wait
	// 2 * lease time for DDL operation.
	if localstore.IsLocalStore(store) {
		lease = 0
	}
	d, err = domain.NewDomain(store, lease)
	if err != nil {
		return nil, errors.Trace(err)
	}
	dm.domains[key] = d
	return
}

func (dm *domainMap) Get(store kv.Storage) (d *domain.Domain, err error) {
	key := store.UUID()
	dm.mu.Lock()
	defer dm.mu.Unlock()
	d = dm.domains[key]
	if d != nil {
		return
	}

	lease := time.Duration(0)
	if !localstore.IsLocalStore(store) {
		lease = schemaLease
	}
	d, err = domain.NewDomain(store, lease)
	if err != nil {
		return nil, errors.Trace(err)
	}
	dm.domains[key] = d
	return
}

// runInBootstrapSession create a special session for boostrap to run.
// If no bootstrap and storage is remote, we must use a little lease time to
// bootstrap quickly, after bootstrapped, we will reset the lease time.
// TODO: Using a bootstap tool for doing this may be better later.
func runInBootstrapSession(store kv.Storage, bootstrap func(Session)) {
	saveLease := schemaLease
	if !localstore.IsLocalStore(store) {
		schemaLease = chooseMinLease(schemaLease, 100*time.Millisecond)
	}
	s, err := createSession(store)
	if err != nil {
		// Bootstrap fail will cause program exit.
		log.Fatal(errors.ErrorStack(err))
	}
	schemaLease = saveLease

	s.SetValue(context.Initing, true)
	bootstrap(s)
	finishBootstrap(store)
	s.ClearValue(context.Initing)

	domain := sessionctx.GetDomain(s)
	domain.Close()
	domap.Delete(store)
}

func (dm *domainMap) Get(store kv.Storage) (d *domain.Domain, err error) {
	key := store.UUID()
	dm.mu.Lock()
	defer dm.mu.Unlock()
	d = dm.domains[key]
	if d != nil {
		return
	}

	lease := time.Duration(0)
	if !localstore.IsLocalStore(store) {
		lease = schemaLease
	}
	err = util.RunWithRetry(defaultMaxRetries, retryInterval, func() (retry bool, err1 error) {
		d, err1 = domain.NewDomain(store, lease)
		return true, errors.Trace(err1)
	})
	if err != nil {
		return nil, errors.Trace(err)
	}
	dm.domains[key] = d
	return
}