gms

var CreateAccounts = polardbxreconcile.NewStepBinder("CreateAccounts",
	func(rc *polardbxreconcile.Context, flow control.Flow) (reconcile.Result, error) {
		polardbx := rc.MustGetPolarDBX()

		mgr, err := rc.GetPolarDBXGMSManager()
		if err != nil {
			return flow.Error(err, "Unable to get GMS manager.")
		}

		accountSecret, err := rc.GetPolarDBXSecret(convention.SecretTypeAccount)
		if err != nil {
			return flow.Error(err, "Unable to get account secret.")
		}

		privileges := polardbx.Spec.Privileges
		rootAccountDefined := false
		for _, priv := range privileges {
			if priv.Username == convention.RootAccount {
				rootAccountDefined = true
				priv.Type = polardbxv1polardbx.Super
			}
			passwd, ok := accountSecret.Data[priv.Username]
			if !ok {
				flow.Logger().Info("Ignore account " + priv.Username)
				continue
			}

			privType := gms.GrantCustomPrivilege
			switch priv.Type {
			case polardbxv1polardbx.Super:
				privType = gms.GrantSuperPrivilege
			case polardbxv1polardbx.ReadWrite:
				privType = gms.GrantReadWritePrivilege
			case polardbxv1polardbx.ReadOnly:
				privType = gms.GrantReadOnlyPrivilege
			case polardbxv1polardbx.DDLOnly:
				privType = gms.GrantDdlPrivilege
			case polardbxv1polardbx.DMLOnly:
				privType = gms.GrantReadWritePrivilege
			}

			err := mgr.CreateDBAccount(priv.Username, string(passwd), &gms.GrantOption{
				Type: privType,
			})
			if err != nil {
				return flow.Error(err, "Unable to create account.", "username", priv.Username)
			}
		}

		if !rootAccountDefined {
			err := mgr.CreateDBAccount(convention.RootAccount,
				string(accountSecret.Data[convention.RootAccount]),
				&gms.GrantOption{Type: gms.GrantSuperPrivilege})
			if err != nil {
				return flow.Error(err, "Unable to create account.", "username", convention.RootAccount)
			}
		}

		return flow.Continue("Accounts created.")
	},
)

var DisableTrailingDNs = polardbxreconcile.NewStepBinder("DisableTrailingDNs",
	func(rc *polardbxreconcile.Context, flow control.Flow) (reconcile.Result, error) {
		polardbx := rc.MustGetPolarDBX()
		topology := polardbx.Status.SpecSnapshot.Topology
		replicas := topology.Nodes.DN.Replicas

		dnStores, err := rc.GetOrderedDNList()
		if err != nil {
			return flow.Error(err, "Unable to get xstores of DN.")
		}
		trailingStores := make([]*polardbxv1.XStore, 0)
		for i, xstore := range dnStores {
			if i >= int(replicas) {
				if controllerutil.ContainsFinalizer(xstore, polardbxmeta.Finalizer) {
					trailingStores = append(trailingStores, xstore)
				}
			}
		}

		if len(trailingStores) == 0 {
			return flow.Pass()
		}

		storageInfos, err := transformIntoStorageInfos(rc, trailingStores)
		if err != nil {
			return flow.Error(err, "Unable to transform xstores into storage infos.")
		}

		mgr, err := rc.GetPolarDBXGMSManager()
		if err != nil {
			return flow.Error(err, "Unable to get GMS manager.")
		}
		err = mgr.DisableStorageNodes(storageInfos...)
		if err != nil {
			return flow.Error(err, "Unable to enable trailing storage nodes in GMS.")
		}

		for _, xstore := range trailingStores {
			controllerutil.RemoveFinalizer(xstore, polardbxmeta.Finalizer)
			err := rc.Client().Update(rc.Context(), xstore)
			if err != nil {
				return flow.Error(err, "Unable to remove finalizer from xstore.", "xstore", xstore.Name)
			}
		}

		return flow.Continue("Trailing DNs are disabled!")
	},
)

var EnableDNs = polardbxreconcile.NewStepBinder("EnableDNs",
	func(rc *polardbxreconcile.Context, flow control.Flow) (reconcile.Result, error) {
		polardbx := rc.MustGetPolarDBX()
		topology := polardbx.Status.SpecSnapshot.Topology
		replicas := topology.Nodes.DN.Replicas

		dnStores, err := rc.GetOrderedDNList()
		if err != nil {
			return flow.Error(err, "Unable to get xstores of DN.")
		}

		storageInfos, err := transformIntoStorageInfos(rc, dnStores[:replicas])
		if err != nil {
			return flow.Error(err, "Unable to transform xstores into storage infos.")
		}

		mgr, err := rc.GetPolarDBXGMSManager()
		if err != nil {
			return flow.Error(err, "Unable to get GMS manager.")
		}

		err = mgr.EnableStorageNodes(storageInfos...)
		if err != nil {
			return flow.Error(err, "Unable to enable storage nodes in GMS.")
		}
		return flow.Continue("DNs are enabled!")
	},
)

var InitializeSchemas = polardbxreconcile.NewStepBinder("InitializeSchemas",
	func(rc *polardbxreconcile.Context, flow control.Flow) (reconcile.Result, error) {
		mgr, err := rc.GetPolarDBXGMSManager()
		if err != nil {
			return flow.Error(err, "Unable to get GMS manager.")
		}

		initialized, err := mgr.IsMetaDBInitialized()
		if err != nil {
			return flow.Error(err, "Unable to determine the status of GMS schemas.")
		}

		if !initialized {
			err = mgr.InitializeMetaDB()
			if err != nil {
				return flow.Error(err, "Unable to initialize GMS schemas.")
			}
		}

		return flow.Continue("GMS schemas initialized.")
	},
)

var LockReadWrite = polardbxreconcile.NewStepBinder("LockReadWrite",
	func(rc *polardbxreconcile.Context, flow control.Flow) (reconcile.Result, error) {
		mgr, err := rc.GetPolarDBXGMSManager()
		if err != nil {
			return flow.Error(err, "Unable to get GMS manager.")
		}
		err = mgr.Lock()
		if err != nil {
			return flow.Error(err, "Unable to lock.")
		}
		return flow.Continue("Locked.")
	},
)

var RestoreSchemas = polardbxreconcile.NewStepBinder("RestoreSchemas",
	func(rc *polardbxreconcile.Context, flow control.Flow) (reconcile.Result, error) {
		mgr, err := rc.GetPolarDBXGMSManager()
		if err != nil {
			return flow.Error(err, "Unable to get GMS manager.")
		}

		restored, err := mgr.IsGmsSchemaRestored()
		if err != nil {
			return flow.Error(err, "Unable to determine the status of GMS schemas.")
		}

		if !restored {
			err = mgr.RestoreSchemas("fake")
			if err != nil {
				return flow.Error(err, "Unable to restore GMS schemas.")
			}
		}

		return flow.Continue("GMS schemas restored.")
	},
)

var UnlockReadWrite = polardbxreconcile.NewStepBinder("UnlockReadWrite",
	func(rc *polardbxreconcile.Context, flow control.Flow) (reconcile.Result, error) {
		mgr, err := rc.GetPolarDBXGMSManager()
		if err != nil {
			return flow.Error(err, "Unable to get GMS manager.")
		}
		err = mgr.Unlock()
		if err != nil {
			return flow.Error(err, "Unable to unlock.")
		}
		return flow.Continue("Unlocked.")
	},
)