package tidbmanager

import (
	"errors"
	"strconv"
	"time"

	"github.com/tidbcloud/aws-shared-provider/pkg/metrics"
	"github.com/tidbcloud/aws-shared-provider/pkg/utils"
	"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) startTokenWatch() error {
	logger := t.WithName("tokenWatch")
LOOP:
	logger.Info("start watch etcd", "revision", t.tokenRevision, "prefix", t.etcdTokenPrefix)
	metrics.WatchLoopCounter.WithLabelValues(t.tidbPoolNamespace).Inc()
	wch := t.etcdWatcher.Watch(t.ctx, t.etcdTokenPrefix, clientv3.WithPrefix(), clientv3.WithRev(t.tokenRevision+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, it will case stop process
				return err
			}
			logger.Info("loop after 2s ...")

			time.Sleep(2 * time.Second) //nolint: gomnd // not need
			goto LOOP
		}

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

			token, keyType := utils.ParseTokenFromKey(t.etcdTokenPrefix, string(kv.Key))
			metrics.EventCounter.WithLabelValues(evType.String(), keyType.String(), t.tidbPoolNamespace).Inc()
			if token == "" {
				logger.Info("empty token", "key", kv.Key)
				continue
			}

			switch keyType {
			case utils.KeyEndpointsType:
				switch evType {
				case mvccpb.DELETE:
					t.removeTokenHandler(token, kv.ModRevision)
				case mvccpb.PUT:
					t.endpointsHandler(token, string(kv.Value), kv.ModRevision)
				default:
					logger.Info("ignore this evType", "token", token, "evType", evType)
				}
			case utils.KeyReplicasType:
				switch evType {
				case mvccpb.DELETE:
					t.removeReplicasHandler(token)
				case mvccpb.PUT:
					replicas, err := strconv.Atoi(string(kv.Value))
					if err != nil {
						logger.Error(err, "Atoi Replicas", "token", token, "replicas", kv.Value)
						continue
					}
					t.replicasHandler(token, replicas, kv.ModRevision)
				default:
					logger.Info("ignore this evType", "token", token, "evType", evType)
				}
			default:
				logger.Info("unknown keyType", "keyType", keyType.String(), "key", kv.Key)
			}
		}
		t.syncCacheSizeMetric()
	}

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