instance

var CheckTopologySpec = plugin.NewStepBinder(galaxy.Engine, "CheckTopologySpec",
	func(rc *xstorev1reconcile.Context, flow control.Flow) (reconcile.Result, error) {
		xstore := rc.MustGetXStore()

		err := checkTopologySpec(rc.MustGetXStore())
		if err != nil {
			xstore.Status.Phase = polardbxv1xstore.PhaseFailed
			return flow.Error(err, "Check topology failed. Transfer phase into Failed.")
		}
		return flow.Pass()
	},
)

var CreateOrUpdateConfigMaps = plugin.NewStepBinder(galaxy.Engine, "CreateOrUpdateConfigMaps",
	func(rc *xstorev1reconcile.Context, flow control.Flow) (reconcile.Result, error) {
		xstore := rc.MustGetXStore()

		for _, cmType := range []convention.ConfigMapType{
			convention.ConfigMapTypeConfig,
			convention.ConfigMapTypeShared,
			convention.ConfigMapTypeTask,
		} {
			loopFlow := flow.WithLoggerValues("configmap-type", cmType)

			cm, err := rc.GetXStoreConfigMap(cmType)
			if client.IgnoreNotFound(err) != nil {
				return loopFlow.Error(err, "Unable to get configmap.")
			}

			if cm == nil {
				cm, err = factory.NewConfigMap(rc, xstore, cmType)
				if err != nil {
					return loopFlow.Error(err, "Unable to construct configmap.")
				}

				if err := rc.SetControllerRefAndCreate(cm); err != nil {
					return loopFlow.Error(err, "Unable to create configmap.")
				}
				loopFlow.Logger().Info("ConfigMap is created.")
			} else {
				outdated, err := convention.IsGenerationOutdated(xstore, cm)
				if err != nil {
					return loopFlow.Error(err, "Unable to resolve generation.")
				}

				if outdated {
					loopFlow.Logger().Info("ConfigMap is outdated, try update.")
					newCm, err := factory.NewConfigMap(rc, xstore, cmType)
					if err != nil {
						return loopFlow.Error(err, "Unable to construct configmap.")
					}
					err = rc.SetControllerRef(newCm)
					if err != nil {
						return loopFlow.Error(err, "Unable to set controller reference.")
					}
					if err := rc.Client().Update(rc.Context(), newCm); err != nil {
						return loopFlow.Error(err, "Unable to update configmap.")
					}
				}
			}
		}

		return flow.Pass()
	},
)

var CreatePodsAndHeadlessServices = xstoreinssteps.CreatePodsAndHeadlessServicesWithExtraFactory(&galaxyfactory.ExtraPodFactoryGalaxy{
	Delegate: &factory.DefaultExtraPodFactory{},
})

var CreateServices = plugin.NewStepBinder(galaxy.Engine, "CreateServices",
	func(rc *xstorev1reconcile.Context, flow control.Flow) (reconcile.Result, error) {
		xstore := rc.MustGetXStore()

		for _, serviceType := range []convention.ServiceType{
			convention.ServiceTypeReadWrite,
			convention.ServiceTypeReadOnly,
			convention.ServiceTypeMetrics,
		} {
			svc, err := rc.GetXStoreService(serviceType)
			if client.IgnoreNotFound(err) != nil {
				return flow.Error(err, "Unable to get service.", "service-type", serviceType)
			}

			if svc == nil {
				svc = factory.NewService(xstore, serviceType)
				err := rc.SetControllerRefAndCreate(svc)
				if err != nil {
					return flow.Error(err, "Unable to create service.", "service-type", serviceType)
				}
			}
		}

		return flow.Continue("Services all ready.")
	},
)

var DummyReconcileConsensusRoleLabels = plugin.NewStepBinder(galaxy.Engine, "ReconcileConsensusRoleLabels",
	func(rc *xstorev1reconcile.Context, flow control.Flow) (reconcile.Result, error) {
		xstore := rc.MustGetXStore()

		pods, err := rc.GetXStorePods()
		if err != nil {
			return flow.Error(err, "Unable to get pods.")
		}

		candidatePods := k8shelper.FilterPodsBy(pods, func(pod *corev1.Pod) bool {
			return xstoremeta.IsPodRoleCandidate(pod)
		})

		if len(candidatePods) > 1 {
			return flow.Error(errors.New("must have only one candidate"), "Found multiple candidate pods, which is not supported yet...")
		}

		currentLeader := xstore.Status.LeaderPod

		candidatePods = k8shelper.FilterPodsBy(candidatePods, k8shelper.IsPodReady)
		if len(candidatePods) > 0 {
			candidatePod := &candidatePods[0]
			currentLeader = candidatePod.Name
			if candidatePod.Labels[xstoremeta.LabelRole] != xstoremeta.RoleLeader {
				candidatePod.Labels[xstoremeta.LabelRole] = xstoremeta.RoleLeader
				err := rc.Client().Update(rc.Context(), candidatePod)
				if err != nil {
					flow.Logger().Error(err, "Unable to reconcile label of node role",
						"pod", candidatePod.Name, "role", xstoremeta.RoleLeader)
				}
			}
		} else {
			currentLeader = ""
		}

		if len(currentLeader) == 0 {
			xstore.Status.LeaderPod = ""

			rc.UpdateXStoreCondition(&polardbxv1xstore.Condition{
				Type:    polardbxv1xstore.LeaderReady,
				Status:  corev1.ConditionFalse,
				Reason:  "LeaderNotFound",
				Message: "Leader not found",
			})

			return flow.Continue("Leader not found!")
		} else if currentLeader != xstore.Status.LeaderPod {
			xstore.Status.LeaderPod = currentLeader

			rc.UpdateXStoreCondition(&polardbxv1xstore.Condition{
				Type:    polardbxv1xstore.LeaderReady,
				Status:  corev1.ConditionTrue,
				Reason:  "LeaderFound",
				Message: "Leader found: " + currentLeader,
			})

			return flow.Continue("Leader changed!", "leader-pod", currentLeader)
		} else {
			xstore.Status.LeaderPod = currentLeader

			rc.UpdateXStoreCondition(&polardbxv1xstore.Condition{
				Type:    polardbxv1xstore.LeaderReady,
				Status:  corev1.ConditionTrue,
				Reason:  "LeaderFound",
				Message: "Leader found: " + currentLeader,
			})

			return flow.Continue("Leader not changed.", "leader-pod", currentLeader)
		}
	})

var SetSuperReadOnlyOnAllCandidates = plugin.NewStepBinder(galaxy.Engine, "SetSuperReadOnlyOnAllCandidates",
	func(rc *xstorereconcile.Context, flow control.Flow) (reconcile.Result, error) {
		pods, err := rc.GetXStorePods()
		if err != nil {
			return flow.Error(err, "Unable to get pods.")
		}

		candidatePods := k8shelper.FilterPodsBy(pods, func(pod *corev1.Pod) bool {
			return xstoremeta.IsPodRoleCandidate(pod)
		})

		for _, pod := range candidatePods {
			err := setSuperReadOnly(rc, flow.Logger(), &pod)
			if err != nil {
				return flow.Error(err, "Unable to set super-read-only on pod.", "pod", pod.Name)
			}
		}

		return flow.Pass()
	},
)

var UnsetSuperReadOnlyOnAllCandidates = plugin.NewStepBinder(galaxy.Engine, "UnsetSuperReadOnlyOnAllCandidates",
	func(rc *xstorereconcile.Context, flow control.Flow) (reconcile.Result, error) {
		pods, err := rc.GetXStorePods()
		if err != nil {
			return flow.Error(err, "Unable to get pods.")
		}

		candidatePods := k8shelper.FilterPodsBy(pods, func(pod *corev1.Pod) bool {
			return xstoremeta.IsPodRoleCandidate(pod)
		})

		for _, pod := range candidatePods {
			err := unsetSuperReadOnly(rc, flow.Logger(), &pod)
			if err != nil {
				return flow.Error(err, "Unable to unset super-read-only on pod.", "pod", pod.Name)
			}
		}

		return flow.Pass()
	},
)