package tests

import (
	"bytes"
	"context"
	"flag"
	"fmt"
	"math/rand"
	"net"
	"net/url"
	"os"
	"path"
	"path/filepath"
	"strings"
	"testing"
	"time"

	"github.com/pingcap/log"
	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/ffs/client"
	"github.com/tidbcloud/ffs/config"
	"github.com/tidbcloud/ffs/pkg/v1/proto"
	"github.com/tidbcloud/ffs/realm"
	"github.com/tidbcloud/ffs/universe"
	"github.com/tidbcloud/ffs/util/netutil"
	"github.com/tidbcloud/ffs/zone"
	"go.etcd.io/etcd/clientv3"
	"go.etcd.io/etcd/embed"
	"google.golang.org/grpc"
)

const (
	universeNum    = 3
	zoneName       = "zone-1"
	zoneNum        = 3
	realmNum       = 3
	etcdClientPort = "12379"
)

var (
	Endpoints     = flag.String("endpoints", "127.0.0.1:"+etcdClientPort, "")
	ClientTempDir = flag.String("client", "", "")
	clientNum     = flag.Int("client-num", 5, "")
)

type Cluster struct {
	Etcd           *embed.Etcd
	StartEtcd      func() (*embed.Etcd, func())
	StopEtcd       func()
	EtcdDir        string
	Universes      []*universe.UniverseServer
	StartUniverses []func() (*universe.UniverseServer, func())
	StopUniverses  []func()
	Zones          []*zone.ZoneServer
	StartZones     []func() (*zone.ZoneServer, func())
	StopZones      []func()
	Realms         []*realm.RealmServer
	StartRealms    []func() (*realm.RealmServer, func())
	StopRealms     []func()
	RealmDirs      []string
	Clients        []*client.Server
	StartClients   []func() (*client.Server, func())
	StopClients    []func()
	ClientDirs     []string
	ctx            context.Context
	cancel         context.CancelFunc
}

func NewCluster(t *testing.T) *Cluster {
	cluster := &Cluster{}
	cluster.ctx, cluster.cancel = context.WithCancel(context.Background())

	etcdDir, err := os.MkdirTemp("", "etcd")
	require.NoError(t, err)
	cluster.EtcdDir = etcdDir
	cluster.StartEtcd = func() (*embed.Etcd, func()) {
		return runEtcd(etcdDir, t)
	}

	cluster.StartUniverses = make([]func() (*universe.UniverseServer, func()), universeNum)
	cluster.StopUniverses = make([]func(), universeNum)
	cluster.Universes = make([]*universe.UniverseServer, universeNum)
	var universeAddrs = make([]string, universeNum)
	for i := 0; i < universeNum; i++ {
		addr := fmt.Sprintf("127.0.0.1:%d", 11000+i)
		cluster.StartUniverses[i] = func() (*universe.UniverseServer, func()) {
			return runUniverse(addr, t)
		}
		universeAddrs[i] = addr
	}
	var universeAddr = strings.Join(universeAddrs, ",")
	cluster.StartZones = make([]func() (*zone.ZoneServer, func()), zoneNum)
	cluster.StopZones = make([]func(), zoneNum)
	cluster.Zones = make([]*zone.ZoneServer, zoneNum)
	for i := 0; i < zoneNum; i++ {
		addr := fmt.Sprintf("127.0.0.1:%d", 12000+i)
		cluster.StartZones[i] = func() (*zone.ZoneServer, func()) {
			return runZone(universeAddr, addr, zoneName, t)
		}
	}

	cluster.RealmDirs = make([]string, realmNum)
	cluster.StartRealms = make([]func() (*realm.RealmServer, func()), realmNum)
	cluster.StopRealms = make([]func(), realmNum)
	cluster.Realms = make([]*realm.RealmServer, realmNum)
	for i := 0; i < realmNum; i++ {
		addr := fmt.Sprintf("127.0.0.1:%d", 13000+i)
		name := fmt.Sprintf("realm-%d", i)
		realmDir, err := os.MkdirTemp("", name)
		require.NoError(t, err)
		log.S().Infof("realm dir %s", realmDir)
		cluster.RealmDirs[i] = realmDir
		cluster.StartRealms[i] = func() (*realm.RealmServer, func()) {
			return runRealm(universeAddr, zoneName, addr, name, realmDir, t)
		}
	}
	clientNum := *clientNum
	cluster.Clients = make([]*client.Server, clientNum)
	cluster.StartClients = make([]func() (*client.Server, func()), clientNum)
	cluster.StopClients = make([]func(), clientNum)
	cluster.ClientDirs = make([]string, clientNum)
	for i := 0; i < clientNum; i++ {
		name := fmt.Sprintf("ffs-client-%d", i)
		var clientDir string
		var err error
		if len(*ClientTempDir) > 0 {
			clientDir = path.Join(*ClientTempDir, name)
			err = os.MkdirAll(clientDir, os.ModePerm)
			require.NoError(t, err)
		} else {
			clientDir, err = os.MkdirTemp("", name)
			require.NoError(t, err)
		}
		log.S().Infof("client dir %s", clientDir)
		cluster.ClientDirs[i] = clientDir
		cluster.StartClients[i] = func() (*client.Server, func()) {
			return runClient(universeAddr, zoneName, clientDir, t)
		}
	}
	return cluster
}

func (c *Cluster) Start() {
	log.S().Infof("Start etcd")
	c.Etcd, c.StopEtcd = c.StartEtcd()
	for i := 0; i < len(c.StartUniverses); i++ {
		c.StartUniverse(i)
	}
	for i := 0; i < len(c.StartZones); i++ {
		c.StartZone(i)
	}
	for i := 0; i < len(c.StartRealms); i++ {
		c.StartRealm(i)
	}
	for i := 0; i < len(c.StartClients); i++ {
		c.StartClient(i)
	}
}

func (c *Cluster) Stop() {
	c.cancel()
	for i := 0; i < len(c.StopClients); i++ {
		stop := c.StopClients[i]
		if stop != nil {
			log.S().Infof("Stop client %d", i)
			stop()
		}
	}
	for i := 0; i < len(c.StopRealms); i++ {
		stop := c.StopRealms[i]
		if stop != nil {
			log.S().Infof("Stop realm %d", i)
			stop()
		}
	}
	for i := 0; i < len(c.StopZones); i++ {
		stop := c.StopZones[i]
		if stop != nil {
			log.S().Infof("Stop zone %d", i)
			stop()
		}
	}
	for i := 0; i < len(c.StopUniverses); i++ {
		stop := c.StopUniverses[i]
		if stop != nil {
			log.S().Infof("Stop universe %d", i)
			stop()
		}
	}
	stop := c.StopEtcd
	if stop != nil {
		log.S().Infof("Stop etcd")
		stop()
	}
	for i := 0; i < len(c.ClientDirs); i++ {
		dir := c.ClientDirs[i]
		log.S().Infof("remove client dir %v", dir)
		os.RemoveAll(dir)
	}
	for i := 0; i < len(c.RealmDirs); i++ {
		dir := c.RealmDirs[i]
		log.S().Infof("remove realm dir %v", dir)
		os.RemoveAll(dir)
	}
	log.S().Infof("remove etcd dir %v", c.EtcdDir)
	os.RemoveAll(c.EtcdDir)
}

func (c *Cluster) Done() <-chan struct{} {
	return c.ctx.Done()
}

func (c *Cluster) StartUniverse(idx int) {
	log.S().Infof("Start universe %d", idx)
	c.Universes[idx], c.StopUniverses[idx] = c.StartUniverses[idx]()
}

func (c *Cluster) StartZone(idx int) {
	log.S().Infof("Start zone %d", idx)
	c.Zones[idx], c.StopZones[idx] = c.StartZones[idx]()
}

func (c *Cluster) StartRealm(idx int) {
	log.S().Infof("Start realm %d", idx)
	c.Realms[idx], c.StopRealms[idx] = c.StartRealms[idx]()
}

func (c *Cluster) StartClient(idx int) {
	log.S().Infof("Start client %d", idx)
	c.Clients[idx], c.StopClients[idx] = c.StartClients[idx]()
}

func runClient(UniverseAddr, ZoneName, target string, t *testing.T) (*client.Server, func()) {
	var (
		tenant        = "tenant"
		volume        = "volume"
		merge         = false
		cacheRoot     = ""
		cacheCapacity = 1073741824
		keepCache     = true
		asyncWrite    = false
		uid           = 0
		gid           = 0
	)
	target, err := filepath.Abs(target)
	require.NoErrorf(t, err, "Failed to get abs file path:%v", target)
	var cacheDir string
	if len(cacheRoot) > 0 {
		cacheDir, err = filepath.Abs(cacheRoot)
		require.NoErrorf(t, err, "Failed to get abs cache path:%v", cacheRoot)
		os.RemoveAll(cacheDir)
		err = os.MkdirAll(cacheDir, os.ModePerm)
		require.NoErrorf(t, err, "Failed to mkdir cache path:%v", cacheDir)
	}
	options := client.ClientOptions{
		Universe:   UniverseAddr,
		Zone:       ZoneName,
		Tenant:     tenant,
		Volume:     volume,
		Target:     target,
		Merged:     merge,
		AsyncWrite: asyncWrite,
		CacheDir:   cacheDir,
		CacheCap:   cacheCapacity,
		KeepCache:  keepCache,
		Uid:        uint32(uid),
		Gid:        uint32(gid),
	}
	server := client.MustMount(&options)
	log.S().Infof("Mounted to %q.", target)
	return server, func() {
		server.Stop()
		server.Unmount()
	}
}

func runRealm(UniverseAddr, ZoneName, RealmAddr, RealmName, realmDir string, t *testing.T) (*realm.RealmServer, func()) {
	var (
		name          = RealmName
		advertiseAddr = RealmAddr
	)
	port := netutil.ParsePort(RealmAddr)
	require.Truef(t, len(port) != 0, "failed to parse port: %v", RealmAddr)
	lis, err := net.Listen("tcp", ":"+port)
	require.NoErrorf(t, err, "failed to listen: %v", err)
	log.S().Infof("server listening at %v", lis.Addr())

	s := grpc.NewServer(
		grpc.MaxRecvMsgSize(config.GrpcMaxMsgSize),
		grpc.MaxSendMsgSize(config.GrpcMaxMsgSize),
	)
	realmSvr := realm.NewRealmServer(UniverseAddr, ZoneName, name, advertiseAddr, realmDir, "")
	proto.RegisterRealmServiceServer(s, realmSvr)
	proto.RegisterRealmControlServiceServer(s, realmSvr)

	go s.Serve(lis)
	return realmSvr, func() {
		realmSvr.Stop()
		s.Stop()
		lis.Close()
	}
}

func runZone(UniverseAddr, ZoneAddr, ZoneName string, t *testing.T) (*zone.ZoneServer, func()) {
	var (
		name          = ZoneName
		advertiseAddr = ZoneAddr
	)
	storeClient, err := clientv3.New(clientv3.Config{
		Endpoints:   strings.Split(*Endpoints, ","),
		DialTimeout: 5 * time.Second,
	})
	require.NoErrorf(t, err, "failed to new etcd client: %v", err)

	port := netutil.ParsePort(ZoneAddr)
	require.Truef(t, len(port) != 0, "failed to parse port: %v", ZoneAddr)
	lis, err := net.Listen("tcp", ":"+port)
	require.NoErrorf(t, err, "failed to listen: %v", err)
	log.S().Infof("server listening at %v", lis.Addr())

	s := grpc.NewServer(
		grpc.MaxRecvMsgSize(config.GrpcMaxMsgSize),
		grpc.MaxSendMsgSize(config.GrpcMaxMsgSize),
	)
	zoneSvr := zone.NewZoneServer(UniverseAddr, name, advertiseAddr, storeClient)
	proto.RegisterZoneServiceServer(s, zoneSvr)

	go s.Serve(lis)
	return zoneSvr, func() {
		zoneSvr.Stop()
		s.Stop()
		lis.Close()
	}
}

func runUniverse(UniverseAddr string, t *testing.T) (*universe.UniverseServer, func()) {
	var (
		advertiseAddr = UniverseAddr
	)
	storeClient, err := clientv3.New(clientv3.Config{
		Endpoints:   strings.Split(*Endpoints, ","),
		DialTimeout: 5 * time.Second,
	})
	require.NoErrorf(t, err, "failed to new etcd client: %v", err)

	port := netutil.ParsePort(UniverseAddr)
	require.Truef(t, len(port) != 0, "failed to parse port: %v", UniverseAddr)
	lis, err := net.Listen("tcp", ":"+port)
	require.NoErrorf(t, err, "failed to listen: %v", err)
	log.S().Infof("server listening at %v", lis.Addr())

	s := grpc.NewServer(
		grpc.MaxRecvMsgSize(config.GrpcMaxMsgSize),
		grpc.MaxSendMsgSize(config.GrpcMaxMsgSize),
	)
	universeSvr := universe.NewUniverseServer(advertiseAddr, storeClient)
	proto.RegisterUniverseServiceServer(s, universeSvr)
	go s.Serve(lis)
	return universeSvr, func() {
		universeSvr.Stop()
		s.Stop()
		lis.Close()
	}
}

func runEtcd(etcdDir string, t *testing.T) (*embed.Etcd, func()) {
	cfg := embed.NewConfig()
	cfg.Name = "ffs"
	cfg.Dir = etcdDir
	cfg.Logger = "zap"
	cfg.LogOutputs = []string{"stdout"}
	cfg.LogLevel = "fatal"
	cfg.LPUrls = parsePeers([]string{"127.0.0.1"})
	cfg.APUrls = parsePeers([]string{"127.0.0.1"})
	cfg.LCUrls = parseClients([]string{"127.0.0.1"})
	cfg.ACUrls = parseClients([]string{"127.0.0.1"})
	cfg.InitialCluster = parseInitialCluster([]string{cfg.Name}, []string{"127.0.0.1"})
	e, err := embed.StartEtcd(cfg)
	require.NoError(t, err)
	select {
	case <-e.Server.ReadyNotify():
		log.S().Info("Server is ready!")
	case <-time.After(time.Minute):
		e.Server.Stop()
		t.Fatalf("Server took too long to start!")
	}
	return e, func() {
		e.Server.Stop()
		e.Close()
	}
}

func parsePeers(eps []string) []url.URL {
	neps := make([]url.URL, len(eps))
	for i, ep := range eps {
		u, err := url.Parse("http://" + ep + ":12380")
		if err != nil {
			return []url.URL{}
		}
		neps[i] = *u
	}
	return neps
}

func parseClients(eps []string) []url.URL {
	neps := make([]url.URL, len(eps))
	for i, ep := range eps {
		u, err := url.Parse("http://" + ep + ":" + etcdClientPort)
		if err != nil {
			return []url.URL{}
		}
		neps[i] = *u
	}
	return neps
}

func parseInitialCluster(names, eps []string) string {
	neps := ""
	for i, ep := range eps {
		neps += names[i] + "=http://" + ep + ":12380"
		if i+1 < len(eps) {
			neps += ","
		}
	}
	return neps
}

func GetRandomIndexes(length int) []int {
	indexes := make([]int, length)
	rand.Shuffle(len(indexes), func(i, j int) {
		indexes[i], indexes[j] = indexes[j], indexes[i]
	})
	return indexes
}

func PrepareFile(filename string, content []byte, t *testing.T) {
	f, err := os.Create(filename)
	require.NoError(t, err)
	n, err := f.Write(content)
	require.NoError(t, err)
	require.Truef(t, n == len(content), "failed to write %d", n)
	err = f.Sync()
	require.NoError(t, err)
	err = f.Close()
	require.NoError(t, err)
}

func AssertFile(filename string, content []byte, t *testing.T) {
	b, err := os.ReadFile(filename)
	require.NoError(t, err)
	require.Truef(t, bytes.Equal(b, content), "file %s content [%s] read [%s]", filename, content, b)
}