package standbywatcher

import (
	"context"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"strings"
	"sync"
	"time"

	"github.com/pingcap/log"
	"github.com/tidbcloud/aws-shared-provider/pkg/tidbworker/resourcemanager/poolmanager/factory"
	rm "github.com/tidbcloud/aws-shared-provider/pkg/tidbworker/resourcemanager/types"
	"go.uber.org/zap"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

const (
	statusPath   = "/tidb-pool/status"
	standbyState = "standby"
)

type standbyWatcherImpl struct {
	manager    rm.PoolManager
	namespace  string
	clusterCli client.Client
	cfg        *rm.StandbyWatcherConfig

	sync.RWMutex
	standbyWorkers map[rm.WorkerPool][]rm.Worker
}

func NewStandbyWatcher(manager rm.PoolManager, cfg *rm.StandbyWatcherConfig) rm.StandbyWatcher {
	return &standbyWatcherImpl{
		manager:        manager,
		namespace:      manager.GetNamespace(),
		clusterCli:     manager.GetClusterClient(),
		cfg:            cfg,
		standbyWorkers: make(map[rm.WorkerPool][]rm.Worker),
	}
}

func (s *standbyWatcherImpl) GetStandbyWorkers(pool rm.WorkerPool) ([]rm.Worker, error) {
	s.RLock()
	defer s.RUnlock()
	return s.standbyWorkers[pool], nil
}

func (s *standbyWatcherImpl) Start(ctx context.Context) error {
	ticker := time.NewTicker(s.cfg.ProbeInterval)
	defer ticker.Stop()
	log.Info("start probing standby workers")
	for {
		select {
		case <-ctx.Done():
			log.Info("stop probing standby workers, context done")
			return nil
		case <-ticker.C:
			if err := s.probeStandbyWorkers(ctx); err != nil {
				log.Error("failed to probe standby workers, will retry later", zap.Error(err))
			}
		}
	}
}

func (s *standbyWatcherImpl) probeStandbyWorkers(ctx context.Context) error {
	standbyWorkers := make(map[rm.WorkerPool][]rm.Worker)
	pools, err := s.manager.GetPools()
	if err != nil {
		log.Error("failed to get pools", zap.Error(err))
		return err
	}
	for pool, pc := range pools {
		standbys, err := s.probePool(ctx, pool, pc.Size)
		if err != nil {
			log.Error("failed to probe pool", zap.Any("pool", pool), zap.Error(err))
			return err
		}
		standbyWorkers[pool] = standbys
	}
	s.Lock()
	defer s.Unlock()
	s.standbyWorkers = standbyWorkers
	return nil
}

func (s *standbyWatcherImpl) probePool(ctx context.Context, pool rm.WorkerPool, size int) ([]rm.Worker, error) {
	standbys := make([]rm.Worker, 0)
	for i := 0; i < size; i++ {
		worker := factory.WorkerHandle(pool, i)
		standby, err := s.probeWorker(ctx, worker)
		if err != nil {
			return nil, err
		}
		if standby {
			standbys = append(standbys, worker)
		}
	}
	return standbys, nil
}

func (s *standbyWatcherImpl) probeWorker(ctx context.Context, worker rm.Worker) (bool, error) {
	url := factory.WorkerStatusEndpoint(worker, s.namespace)
	req, err := s.manager.NewRequest(ctx, http.MethodGet, url, statusPath, nil)
	if err != nil {
		return false, fmt.Errorf("create http request error: %w", err)
	}
	resp, err := s.manager.Do(req)
	if err != nil {
		if strings.Contains(err.Error(), "no such host") {
			log.Info("worker not found, skip checking status", zap.Any("worker", worker))
			return false, nil
		}
		return false, err
	}
	defer resp.Body.Close() //nolint:errcheck // ignore error

	body, err := io.ReadAll(resp.Body)
	if err != nil {
		return false, fmt.Errorf("read body error: %w", err)
	}

	if resp.StatusCode != http.StatusOK {
		return false, fmt.Errorf("status: %s, body: %s", resp.Status, body)
	}
	var status struct {
		State string `json:"state"`
	}
	if err := json.Unmarshal(body, &status); err != nil {
		return false, fmt.Errorf("unmarshal body error: %w", err)
	}

	return status.State == standbyState, nil
}