package k8s

import (
	"context"
	"fmt"
	"path"
	"strings"
	"sync"
	"time"

	"github.com/juju/errors"
	"github.com/ngaut/log"
	agClient "github.com/pingcap/schrodinger/agent/client"
	"github.com/pingcap/schrodinger/agent/core"
	"github.com/pingcap/schrodinger/agent/daemon"
	"github.com/pingcap/schrodinger/pkg/types"
	"github.com/pingcap/schrodinger/util"
	"github.com/pingcap/tidb-operator/pkg/label"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/apimachinery/pkg/util/wait"
	"k8s.io/client-go/kubernetes"
)

// backup tidb cluster physically
func (c *ClusterController) backupCluster(ctx context.Context, tidbClusterBackup *types.TiDBClusterBackup) error {
	log.Infof("namespace: %s, start to get cluster pods", tidbClusterBackup.Namespace)
	pods, err := c.getClusterAllPods(ctx, tidbClusterBackup.Namespace)
	if err != nil {
		log.Errorf("namespace: %s, get cluster pods failed: %v", tidbClusterBackup.Namespace, err)
		return errors.Trace(err)
	}

	log.Infof("namespace: %s, start to generate cluster agent clients", tidbClusterBackup.Namespace)
	agentClients := c.genClusterAgentClients(pods)

	log.Infof("namespace: %s, start to stop cluster", tidbClusterBackup.Namespace)
	if err = c.stopCluster(agentClients); err != nil {
		log.Errorf("namespace: %s, stop cluster failed: %v", tidbClusterBackup.Namespace, err)
		return errors.Trace(err)
	}

	log.Infof("namespace: %s, start to execute backup cluster", tidbClusterBackup.Namespace)
	if err = c.backupClusterAction(tidbClusterBackup, pods); err != nil {
		log.Errorf("namespace: %s, execute backup cluster failed: %v", tidbClusterBackup.Namespace, err)
		return errors.Trace(err)
	}

	log.Infof("namespace: %s, start to restart cluster", tidbClusterBackup.Namespace)
	if err = c.restartCluster(agentClients); err != nil {
		log.Errorf("namespace: %s, restart cluster failed: %v", tidbClusterBackup.Namespace, err)
		return errors.Trace(err)
	}

	log.Infof("namespace: %s, waiting tidb cluster is running", tidbClusterBackup.Namespace)
	if err = c.waitingClusterIsRunning(ctx, tidbClusterBackup.Namespace); err != nil {
		log.Errorf("namespace: %s, tidb cluster is not running, err: %v", tidbClusterBackup.Namespace, err)
		return errors.Trace(err)
	}
	return nil
}

// recover cluster from backup
func (c *ClusterController) recoverCluster(ctx context.Context, tidbClusterBackup *types.TiDBClusterBackup) error {
	log.Infof("namespace: %s, start to get cluster pods", tidbClusterBackup.Namespace)
	pods, err := c.getClusterAllPods(ctx, tidbClusterBackup.Namespace)
	if err != nil {
		log.Errorf("namespace: %s, get cluster pods faield: %v", tidbClusterBackup.Namespace, err)
		return errors.Trace(err)
	}

	log.Infof("namespace: %s, start to generate cluster agent clients", tidbClusterBackup.Namespace)
	agentClients := c.genClusterAgentClients(pods)

	// TODO: need to recover size of pds and tikvs

	log.Infof("namespace: %s, start to stop cluster", tidbClusterBackup.Namespace)
	if err = c.stopCluster(agentClients); err != nil {
		log.Errorf("namespace: %s, stop cluster failed: %v", tidbClusterBackup.Namespace, err)
		return errors.Trace(err)
	}

	// restore action
	if err = c.restoreClusterAction(tidbClusterBackup, pods); err != nil {
		log.Errorf("namespace: %s do restore jobs failed: %v", tidbClusterBackup.Namespace, err)
		return errors.Trace(err)
	}

	log.Infof("namespace: %s, start to restart cluster", tidbClusterBackup.Namespace)
	if err = c.restartCluster(agentClients); err != nil {
		log.Errorf("namespace: %s, restart cluster failed: %v", tidbClusterBackup.Namespace, err)
		return errors.Trace(err)
	}

	log.Infof("namespace: %s, waiting tidb cluster is running", tidbClusterBackup.Namespace)
	if err = c.waitingClusterIsRunning(ctx, tidbClusterBackup.Namespace); err != nil {
		log.Errorf("namespace: %s, tidb cluster is not running, err: %v", tidbClusterBackup.Namespace, err)
		return errors.Trace(err)
	}
	return nil
}

// backup pds and tikvs using tidb-sch-ctl physically
func (c *ClusterController) backupClusterAction(tidbClusterBackup *types.TiDBClusterBackup, pods map[string]*corev1.PodList) error {
	tidbClusterBackup.Spec = types.TiDBClusterBackupSpec{
		Cluster:        tidbClusterBackup.Name,
		PhysicalBackup: &types.PhysicalBackupSpec{},
	}

	// backup pd
	tidbClusterBackup.Status.PhysicalBackupStatus.PDBackupTaskStatuses = make(map[string]*types.BackupTaskStatus, len(pods["pd"].Items))
	if err := c.doBackup(tidbClusterBackup, pods["pd"], label.PDLabelVal, label.MemberIDLabelKey, c.execBackupPD); err != nil {
		log.Errorf("namespace: %s, backup pds failed: %v", tidbClusterBackup.Namespace, err)
		return errors.Trace(err)
	}

	// backup tikv
	tidbClusterBackup.Status.PhysicalBackupStatus.TiKVBackupTaskStatuses = make(map[string]*types.BackupTaskStatus, len(pods["tikv"].Items))
	if err := c.doBackup(tidbClusterBackup, pods["tikv"], label.TiKVLabelVal, label.StoreIDLabelKey, c.execBackupTiKV); err != nil {
		log.Errorf("namespace: %s, backup tikvs failed: %v", tidbClusterBackup.Namespace, err)
		return errors.Trace(err)
	}

	return nil
}

func (c *ClusterController) doBackup(
	tidbClusterBackup *types.TiDBClusterBackup,
	pods *corev1.PodList,
	appLableKey,
	idLabelKey string,
	fn backupFunc,
) error {
	pvs, err := findPVsByLabel(c.K8sCli, tidbClusterBackup.Namespace, appLableKey)
	if err != nil {
		return errors.Trace(err)
	}

	lock := new(sync.RWMutex)
	wg := sync.WaitGroup{}
	errChan := make(chan error, len(pods.Items))
	for _, pod := range pods.Items {
		id, exist := pod.Labels[idLabelKey]
		if !exist {
			return errors.NotFoundf("label %s", idLabelKey)
		}
		pv, ok := findPVByLabelVal(pvs, idLabelKey, id)
		if !ok {
			log.Errorf("namespace: %s, get pv by id %s failed, not found", tidbClusterBackup.Namespace, id)
			return errors.Errorf("namespace: %s, get pv by id %s failed, not found", tidbClusterBackup.Namespace, id)
		}

		wg.Add(1)
		go func(pod corev1.Pod) {
			if err := fn(tidbClusterBackup, pod, pv, id, &wg, lock); err != nil {
				errChan <- err
			}
		}(pod)
	}

	wg.Wait()

	if len(errChan) > 0 {
		return errors.New("fail to exec backup pod")
	}

	return nil
}

type backupFunc func(
	tidbClusterBackup *types.TiDBClusterBackup,
	pod corev1.Pod,
	pv *corev1.PersistentVolume,
	id string,
	wg *sync.WaitGroup,
	lock *sync.RWMutex,
) error

func (c *ClusterController) execBackupPD(
	tidbClusterBackup *types.TiDBClusterBackup,
	pod corev1.Pod,
	pv *corev1.PersistentVolume,
	memberID string,
	wg *sync.WaitGroup,
	lock *sync.RWMutex,
) error {
	backupStatus := getBackupTaskStatus(pod, memberID)

	defer func() {
		lock.Lock()
		tidbClusterBackup.Status.PhysicalBackupStatus.PDBackupTaskStatuses[backupStatus.ID] = backupStatus
		lock.Unlock()
		tidbClusterBackup.Spec.PhysicalBackup.BackupMembers = append(tidbClusterBackup.Spec.PhysicalBackup.BackupStores, backupStatus.ID)
		wg.Done()
	}()

	backupStatus.BackupDir = path.Join(pod.Annotations[HostPathLabelKey], BackupStoreDir, tidbClusterBackup.Name)

	ag := agClient.NewClient(fmt.Sprintf("%s:%d", pod.Status.PodIP, c.ClusterAgentPort))
	srcPath := path.Join(PDDataMountPath, PDStoreDir)
	backupPath := path.Join(PDDataMountPath, BackupStoreDir, tidbClusterBackup.Name)
	cmd := fmt.Sprintf("/physical_backup_pd.sh %s %s", srcPath, backupPath)
	backupStatus.State = types.BackupRunning

	log.Infof("namespace: %v, start to exec [%v] on pod %s", tidbClusterBackup.Namespace, cmd, pod.Name)
	// TODO: need retry
	if output, err := ag.ExecCommand(&core.ShellCommand{Command: cmd}); err != nil {
		backupStatus.State = types.BackupFailed
		backupStatus.Message = fmt.Sprintf("exec [%v] failed: %v", tidbClusterBackup.Namespace, err)
		log.Errorf("namespace: %v, exec [%v] failed, output: %v, error: %v", tidbClusterBackup.Namespace, cmd, output, err)
		return errors.Trace(err)
	}
	backupStatus.State = types.BackupSuccess

	return nil
}

func (c *ClusterController) execBackupTiKV(
	tidbClusterBackup *types.TiDBClusterBackup,
	pod corev1.Pod,
	pv *corev1.PersistentVolume,
	storeID string,
	wg *sync.WaitGroup,
	lock *sync.RWMutex,
) error {
	backupStatus := getBackupTaskStatus(pod, storeID)
	defer func() {
		lock.Lock()
		tidbClusterBackup.Status.PhysicalBackupStatus.TiKVBackupTaskStatuses[backupStatus.ID] = backupStatus
		lock.Unlock()
		tidbClusterBackup.Spec.PhysicalBackup.BackupStores = append(tidbClusterBackup.Spec.PhysicalBackup.BackupStores, backupStatus.ID)
		wg.Done()
	}()

	backupStatus.BackupDir = path.Join(pod.Annotations[HostPathLabelKey], BackupStoreDir, tidbClusterBackup.Name)

	ag := agClient.NewClient(fmt.Sprintf("%s:%d", pod.Status.PodIP, c.ClusterAgentPort))
	srcPath := path.Join(TiKVDataMountPath, TiKVStoreDir)
	backupPath := path.Join(TiKVDataMountPath, BackupStoreDir, tidbClusterBackup.Name)
	cmd := fmt.Sprintf("/physical_backup_tikv.sh %s %s", srcPath, backupPath)
	backupStatus.State = types.BackupRunning

	log.Infof("namespace: %s, start to exec [%v] on pod %s", tidbClusterBackup.Namespace, cmd, pod.Name)
	// TODO: need retry
	if output, err := ag.ExecCommand(&core.ShellCommand{Command: cmd}); err != nil {
		backupStatus.State = types.BackupFailed
		backupStatus.Message = fmt.Sprintf("exec [%v] failed: %v", tidbClusterBackup.Namespace, err)
		log.Errorf("namespace: %s, exec [%v] failed, output: %v, error: %v", tidbClusterBackup.Namespace, cmd, output, err)
		return errors.Trace(err)
	}
	backupStatus.State = types.BackupSuccess

	return nil
}

func findPVsByLabel(k8sCli kubernetes.Interface, namespace, lab string) ([]corev1.PersistentVolume, error) {
	pvLabel := label.New().Namespace(namespace).Instance(namespace).Component(lab)
	option := metav1.ListOptions{
		LabelSelector: labels.SelectorFromSet(pvLabel.Labels()).String(),
	}

	pvList, err := k8sCli.CoreV1().PersistentVolumes().List(option)
	if err != nil {
		return nil, errors.Trace(err)
	}

	return pvList.Items, nil
}

func findPVByLabelVal(pvs []corev1.PersistentVolume, key, val string) (*corev1.PersistentVolume, bool) {
	for _, pv := range pvs {
		if v, ok := pv.Labels[key]; ok && v == val {
			return &pv, true
		}
	}
	return nil, false
}

func getBackupTaskStatus(pod corev1.Pod, id string) *types.BackupTaskStatus {
	return &types.BackupTaskStatus{
		TaskName:   pod.GetName(),
		BeginTime:  time.Now().Format(util.TimeFormat),
		State:      types.BackupPending,
		BackupNode: pod.Spec.NodeName,
		ID:         id,
	}
}

func (c *ClusterController) restartTiKVs(ctx context.Context, namespace string) error {
	tikvPods, err := c.getClusterPodsByLabel(ctx, namespace, label.TiKVLabelVal)
	if err != nil {
		return errors.Trace(err)
	}

	tikvClients := genClients(tikvPods, c.ClusterAgentPort)

	log.Infof("namespace: %s, kill all tikvs", namespace)
	if err = agentDoKill(tikvClients, "tikv"); err != nil {
		return errors.Trace(err)
	}

	log.Infof("namespace: %s, restart all tikvs", namespace)
	if err = agentDoRestart(tikvClients, "tikv"); err != nil {
		return errors.Trace(err)
	}

	return nil
}

func (c *ClusterController) stopTiDBs(ctx context.Context, namespace string) error {
	tidbPods, err := c.getClusterPodsByLabel(ctx, namespace, label.TiDBLabelVal)
	if err != nil {
		return errors.Trace(err)
	}

	tidbClients := genClients(tidbPods, c.ClusterAgentPort)
	if err = agentDoKill(tidbClients, "tidb"); err != nil {
		return errors.Trace(err)
	}

	return nil
}

func (c *ClusterController) startTiDBs(ctx context.Context, namespace string) error {
	tidbPods, err := c.getClusterPodsByLabel(ctx, namespace, label.TiDBLabelVal)
	if err != nil {
		return errors.Trace(err)
	}

	tidbClients := genClients(tidbPods, c.ClusterAgentPort)
	if err = agentDoRestart(tidbClients, "tidb"); err != nil {
		return errors.Trace(err)
	}

	return nil
}

func (c *ClusterController) getClusterAllPods(ctx context.Context, namespace string) (map[string]*corev1.PodList, error) {
	tikvPods, err := c.getClusterPodsByLabel(ctx, namespace, label.TiKVLabelVal)
	if err != nil {
		return nil, errors.Trace(err)
	}

	pdPods, err := c.getClusterPodsByLabel(ctx, namespace, label.PDLabelVal)
	if err != nil {
		return nil, errors.Trace(err)
	}

	tidbPods, err := c.getClusterPodsByLabel(ctx, namespace, label.TiDBLabelVal)
	if err != nil {
		return nil, errors.Trace(err)
	}

	return map[string]*corev1.PodList{
		"tikv": tikvPods,
		"pd":   pdPods,
		"tidb": tidbPods,
	}, nil
}

func (c *ClusterController) genClusterAgentClients(pods map[string]*corev1.PodList) map[string][]*agClient.Client {
	return map[string][]*agClient.Client{
		"tikv": genClients(pods["tikv"], c.ClusterAgentPort),
		"pd":   genClients(pods["pd"], c.ClusterAgentPort),
		"tidb": genClients(pods["tidb"], c.ClusterAgentPort),
	}
}

// use agent to stop pd / tikv
func (c *ClusterController) stopCluster(agentClients map[string][]*agClient.Client) error {
	log.Info("start to stop tidbs")
	if err := agentDoKill(agentClients["tidb"], "tidb"); err != nil {
		return errors.Trace(err)
	}

	log.Info("start to stop pds")
	if err := agentDoKill(agentClients["pd"], "pd"); err != nil {
		return errors.Trace(err)
	}

	log.Info("start to stop tikvs")
	if err := agentDoKill(agentClients["tikv"], "tikv"); err != nil {
		return errors.Trace(err)
	}

	return nil
}

// use agnet to restart pd / tikv / tidb
func (c *ClusterController) restartCluster(agentClients map[string][]*agClient.Client) error {
	log.Info("start to restart pds")
	if err := agentDoRestart(agentClients["pd"], "pd"); err != nil {
		return errors.Trace(err)
	}

	log.Info("sleep one minutes to make sure pd is ok")
	time.Sleep(1 * time.Minute)

	log.Infof("start to restart tikvs")
	if err := agentDoRestart(agentClients["tikv"], "tikv"); err != nil {
		return errors.Trace(err)
	}

	log.Info("sleep one minutes to make sure tikv is ok")
	time.Sleep(1 * time.Minute)

	log.Infof("start to restart tidbs")
	if err := agentDoRestart(agentClients["tidb"], "tidb"); err != nil {
		return errors.Trace(err)
	}
	return nil
}

func (c *ClusterController) waitingClusterIsRunning(ctx context.Context, namespace string) error {
	clu, err := c.ClusterCli.PingcapV1alpha1().TidbClusters(namespace).Get(TiDBClusterNameSuffix, metav1.GetOptions{})
	if err != nil {
		return errors.Trace(err)
	}

	if clu == nil {
		return errors.NotFoundf("namespace %s tidb cluster", namespace)
	}

	cluster := &types.Cluster{
		TidbCluster: clu,
	}

	return c.getClusterAccessInfo(ctx, cluster, true)
}

func (c *ClusterController) waitingRunningPodsEqualSizeByLabel(ctx context.Context, namespace, lab string, expectedSize int32) error {
	err := wait.Poll(10*time.Second, 5*time.Minute, func() (bool, error) {
		listOptions := metav1.ListOptions{
			LabelSelector: labels.SelectorFromSet(label.New().Instance(namespace).Component(lab).Labels()).String(),
		}

		tikvPods, err := c.K8sCli.CoreV1().Pods(namespace).List(listOptions)
		if err != nil {
			return false, nil
		}

		if expectedSize == 0 {
			return len(tikvPods.Items) == 0, nil
		}

		var running int32
		for _, pod := range tikvPods.Items {
			if pod.Status.Phase == corev1.PodRunning {
				containerRunning := 0
				for _, cs := range pod.Status.ContainerStatuses {
					if cs.State.Running != nil {
						containerRunning++
					}
				}
				if containerRunning >= len(pod.Status.ContainerStatuses) {
					running++
				}
			}
		}
		if running == expectedSize {
			return true, nil
		}
		log.Infof("cluster %s except running %s size %d, get %d", namespace, lab, expectedSize, running)
		return false, nil
	})
	return errors.Trace(err)
}

func (c *ClusterController) restoreClusterAction(tidbClusterBackup *types.TiDBClusterBackup, pods map[string]*corev1.PodList) error {
	log.Infof("namespace: %s, restore pds", tidbClusterBackup.Namespace)
	pdDataDir := path.Join(PDDataMountPath, PDStoreDir)
	pdBackupDir := path.Join(PDDataMountPath, BackupStoreDir, tidbClusterBackup.Name)
	if err := c.doRestore(tidbClusterBackup, pods["pd"], pdDataDir, pdBackupDir); err != nil {
		log.Errorf("namespace: %s, restore pds failed: %v", tidbClusterBackup.Namespace, err)
		return errors.Trace(err)
	}

	log.Infof("namespace: %s, start to restore tikvs", tidbClusterBackup.Namespace)
	tikvDataDir := path.Join(TiKVDataMountPath, TiKVStoreDir)
	tikvBackupDir := path.Join(TiKVDataMountPath, BackupStoreDir, tidbClusterBackup.Name)
	if err := c.doRestore(tidbClusterBackup, pods["tikv"], tikvDataDir, tikvBackupDir); err != nil {
		log.Errorf("namespace: %s, restore pds failed: %v", tidbClusterBackup.Namespace, err)
		return errors.Trace(err)
	}
	return nil
}

func (c *ClusterController) doRestore(tidbClusterBackup *types.TiDBClusterBackup, pods *corev1.PodList, dataDir, backupDir string) error {
	wg := sync.WaitGroup{}
	errChan := make(chan error, len(pods.Items))
	for _, pod := range pods.Items {
		wg.Add(1)
		go func(pod corev1.Pod) {
			if err := c.restorePod(tidbClusterBackup, pod, dataDir, backupDir, &wg); err != nil {
				errChan <- err
			}
		}(pod)
	}
	wg.Wait()

	if len(errChan) > 0 {
		return errors.New("fail to exec restore pod")
	}

	return nil
}

func (c *ClusterController) restorePod(tidbClusterBackup *types.TiDBClusterBackup, pod corev1.Pod, dataDir, backupDir string, wg *sync.WaitGroup) error {
	defer wg.Done()

	ag := agClient.NewClient(fmt.Sprintf("%s:%d", pod.Status.PodIP, c.ClusterAgentPort))
	// cmd := fmt.Sprintf("rm -rf /var/lib/tidb/%s/* && cp -rf /var/lib/tidb/backup/%s/* /var/lib/tidb/%s/", dataDir, tidbClusterBackup.Name, dataDir)
	cmd := fmt.Sprintf("rm -rf %s/* && cp -rf %s/* %s/", dataDir, backupDir, dataDir)

	log.Infof("namespace: %s, start to exec [%v] on pod %s", tidbClusterBackup.Namespace, cmd, pod.Name)

	// TODO: need retry
	if output, err := ag.ExecCommand(&core.ShellCommand{Command: cmd}); err != nil {
		log.Errorf("namespace: %s, exec [%v] failed, output: %v, error: %v", tidbClusterBackup.Namespace, cmd, output, err)
		return errors.Trace(err)
	}

	return nil
}

func (c *ClusterController) getClusterPodsByLabel(ctx context.Context, namespace, labelVel string) (*corev1.PodList, error) {
	listOptions := metav1.ListOptions{
		LabelSelector: labels.SelectorFromSet(label.New().Instance(namespace).Component(labelVel).Labels()).String(),
	}

	pods, err := c.K8sCli.CoreV1().Pods(namespace).List(listOptions)
	if err != nil {
		return nil, errors.Trace(err)
	}

	return pods, nil
}

func genHostPath(backupPath string) string {
	paths := strings.Split(backupPath, "/")

	if len(paths) < 3 {
		return ""
	}

	return strings.Join(paths[0:3], "/")
}

func genClients(pods *corev1.PodList, agentPort int) []*agClient.Client {
	var clients []*agClient.Client
	for _, pod := range pods.Items {
		clients = append(clients, agClient.NewClient(fmt.Sprintf("%s:%d", pod.Status.PodIP, agentPort)))
	}
	return clients
}

func agentDoRestart(clients []*agClient.Client, processName string) error {
	for _, cli := range clients {
		err := wait.Poll(20*time.Second, 5*time.Minute, func() (bool, error) {
			if err := cli.RestartProcess(processName); err != nil {
				if strings.Contains(err.Error(), "start process from state [RUNNING]") {
					return true, nil
				}

				log.Errorf("restart %s failed: %v, retry", processName, err)
				return false, nil
			}

			if err := checkProcessState(cli, processName, daemon.ProcStatRunning); err != nil {
				log.Errorf("check %s expect status %s failed, error: %v, retry", processName, daemon.ProcStatRunning, err)
				return false, nil
			}
			return true, nil
		})

		if err != nil {
			return errors.Trace(err)
		}
	}
	return nil
}

func agentDoKill(clients []*agClient.Client, processName string) error {
	for _, cli := range clients {
		err := wait.Poll(10*time.Second, 1*time.Minute, func() (bool, error) {
			if err := cli.StopProcess(processName); err != nil {
				tmpState, err := cli.ProcessState(processName)
				if err != nil {
					log.Errorf("get %s state failed: %v, retry", processName, err)
					return false, nil
				}

				log.Infof("get %s state: %s", processName, tmpState.String())
				if tmpState == daemon.ProcStatKilled ||
					tmpState == daemon.ProcStatErrorExited ||
					tmpState == daemon.ProcStatNormalExited ||
					tmpState == daemon.ProcStatStopped {
					return true, nil
				}
				log.Errorf("stop %s failed: %v, retry", processName, err)
				return false, nil
			}

			if err := checkProcessState(cli, processName, daemon.ProcStatKilled); err != nil {
				return true, errors.Trace(err)
			}
			return true, nil
		})

		if err != nil {
			return errors.Trace(err)
		}
	}
	return nil
}

func checkProcessState(cli *agClient.Client, processName string, state daemon.ProcessState) error {
	err := wait.Poll(10*time.Second, 30*time.Second, func() (bool, error) {
		tmpState, err := cli.ProcessState(processName)
		if err != nil {
			log.Errorf("get %s state failed: %v, retry", processName, err)
			return false, nil
		}

		if tmpState != state {
			log.Errorf("get %s state: %s, expect state: %s, retry", processName,
				tmpState.String(), state.String())
			return false, nil
		}

		return true, nil
	})
	return errors.Trace(err)
}