package tidbmanager

import (
	"errors"
	"time"

	"go.etcd.io/etcd/api/v3/mvccpb"
	"go.etcd.io/etcd/api/v3/v3rpc/rpctypes"
	clientv3 "go.etcd.io/etcd/client/v3"
)

func (t *tidbManager) startToFreeCacheAgainWatch() error {
	logger := t.WithName("toFreeCacheAgainWatch")
LOOP:
	logger.Info("start watch etcd", "revision", t.toFreeCacheAgainRevision, "prefix", t.etcdToFreeCacheAgain)
	wch := t.etcdWatcher.Watch(t.ctx, t.etcdToFreeCacheAgain, clientv3.WithPrefix(), clientv3.WithRev(t.toFreeCacheAgainRevision+1))
	for wresp := range wch {
		if err := wresp.Err(); err != nil {
			logger.Error(err, "watch etcd")
			if errors.Is(err, rpctypes.ErrCompacted) {
				// ErrCompacted can't recover, return to let caller handle
				return err
			}
			logger.Info("loop after 2s ...")
			time.Sleep(2 * time.Second) //nolint:mnd // retry delay
			goto LOOP
		}

		t.toFreeCacheAgainRevision = wresp.Header.GetRevision()
		for _, ev := range wresp.Events {
			evType, kv := ev.Type, ev.Kv
			logger.V(3).Info("etcd event", "type", evType, "kv", kv)

			switch evType {
			case mvccpb.PUT:
				podName := string(kv.Value)
				logger.V(2).Info("try set to free cache again", "pod name", podName)
				t.checkAndSetToFreeCache(t.ctx, podName)
			case mvccpb.DELETE:
				// nothing to do on delete
				logger.Info("key deleted", "key", string(kv.Key))
			default:
				logger.Info("ignore this evType", "evType", evType)
			}
		}
	}

	err := t.ctx.Err()
	logger.Error(err, "stop watch etcd")
	return err
}