package client

import (
	"context"
	"os"
	"os/signal"
	"strings"
	"sync"
	"sync/atomic"
	"syscall"
	"time"

	"github.com/hanwen/go-fuse/v2/fs"
	"github.com/hanwen/go-fuse/v2/fuse"
	"github.com/hslam/lru"
	"github.com/pingcap/log"
	"github.com/tidbcloud/ffs/config"
	"github.com/tidbcloud/ffs/pkg/v1/proto"
	rclient "github.com/tidbcloud/ffs/realm/client"
	"github.com/tidbcloud/ffs/realm/fserror"
	"github.com/tidbcloud/ffs/realm/mergefs"
	uclient "github.com/tidbcloud/ffs/universe/client"
	zclient "github.com/tidbcloud/ffs/zone/client"
)

type Server struct {
	*fuse.Server
	ctx *Context
}

// 180s is enough for an operation to complete
var cacheDuration = config.Timeout

type Context struct {
	uclient    *uclient.Client
	zclient    *zclient.Client
	client     *rclient.Client
	clients    []*rclient.Client
	realms     []*proto.Realm
	index      uint64
	zone       string
	tenant     string
	volume     string
	target     string
	token      string
	ctx        context.Context
	cancel     context.CancelFunc
	merged     bool
	asyncWrite bool
	cacheDir   string
	keepCache  bool
	cache      struct {
		sync.Mutex
		*lru.LRU
	}
	mergedfs struct {
		sync.RWMutex
		entries map[string]*proto.DirEntry
	}
	wg *sync.WaitGroup
}

func (ctx *Context) Client() *rclient.Client {
	if !ctx.merged {
		return ctx.client
	}
	return ctx.clients[ctx.clientIndex()]
}

func (ctx *Context) clientIndex() uint64 {
	return atomic.LoadUint64(&ctx.index) % uint64(len(ctx.clients))
}

func (ctx *Context) Next() {
	atomic.AddUint64(&ctx.index, 1)
}

// ClientOptions defines parameters specified in ffs client.
type ClientOptions struct {
	Universe   string
	Zone       string
	Tenant     string
	Volume     string
	Target     string
	Merged     bool
	AsyncWrite bool
	CacheDir   string
	CacheCap   int
	KeepCache  bool
	Uid        uint32
	Gid        uint32
}

func RemoveMount(options *ClientOptions) {
	if options == nil {
		log.S().Fatal("client options is nil")
	}
	callCtx, cancel := context.WithTimeout(context.Background(), time.Minute)
	defer cancel()

	uclient := uclient.NewClient(strings.Split(options.Universe, ",")...)
	if err := uclient.Start(); err != nil {
		log.S().Fatalf("failed to start universe client: %v", err)
	}
	defer uclient.Stop()
	var z *proto.Zone
	if resp, err := uclient.Client().GetZone(callCtx, &proto.GetZoneReq{Name: options.Zone}); err != nil {
		log.S().Fatalf("failed to get zone: %v", err)
	} else if len(resp.Endpoints) == 0 {
		log.S().Fatalf("failed to get zone %v endpoint", resp.Name)
	} else {
		log.S().Infof("get zone name: %s endpoints: %s", resp.Name, resp.Endpoints)
		z = resp
	}
	zclient := zclient.NewClient(z.Endpoints...)
	if err := zclient.Start(); err != nil {
		log.S().Fatalf("failed to start zone client: %v", err)
	}
	defer zclient.Stop()
	var realms []*proto.Realm
	if resp, err := zclient.Client().Locate(callCtx, &proto.LocateReq{
		Tenant:     options.Tenant,
		MountPoint: options.Volume,
		Create:     false,
	}); err != nil {
		log.S().Fatalf("failed to get realm: %v", err)
	} else if len(resp.Realms) == 0 {
		return
	} else {
		log.S().Infof("get realms %v", resp.Realms)
		realms = resp.Realms
	}
	for i := range realms {
		rclient := rclient.NewClient(realms[i].Address)
		if err := rclient.Start(); err != nil {
			log.S().Fatalf("failed to start realm client: %v", err)
		}
		defer rclient.Stop()
		if _, err := rclient.Client().Cleanup(callCtx, &proto.CleanupReq{
			Tenant: options.Tenant,
			Name:   options.Volume,
		}); err != nil {
			log.S().Fatalf("failed to cleanup in realm: %v", err)
		}
	}
	if _, err := zclient.Client().RemoveMount(callCtx, &proto.RemoveMountReq{
		Tenant:     options.Tenant,
		MountPoint: options.Volume,
	}); err != nil {
		log.S().Fatalf("failed to get realm: %v", err)
	}
}

func MustMount(options *ClientOptions) *Server {
	if options == nil {
		log.S().Fatal("client options is nil")
	}
	callCtx, cancel := context.WithTimeout(context.Background(), time.Minute)
	defer cancel()

	uclient := uclient.NewClient(strings.Split(options.Universe, ",")...)
	if err := uclient.Start(); err != nil {
		log.S().Fatalf("failed to start universe client: %v", err)
	}
	if resp, err := uclient.Client().CreateTenant(callCtx,
		&proto.CreateTenantReq{Tenant: &proto.Tenant{Name: options.Tenant}},
	); err != nil {
		log.S().Fatalf("failed to create tenant: %v", err)
	} else {
		log.S().Infof("create tenant name: %s", resp.Name)
	}
	var z *proto.Zone
	if resp, err := uclient.Client().GetZone(callCtx, &proto.GetZoneReq{Name: options.Zone}); err != nil {
		log.S().Fatalf("failed to get zone: %v", err)
	} else if len(resp.Endpoints) == 0 {
		log.S().Fatalf("failed to get zone %v endpoint", resp.Name)
	} else {
		log.S().Infof("get zone name: %s endpoints: %s", resp.Name, resp.Endpoints)
		z = resp
	}
	zclient := zclient.NewClient(z.Endpoints...)
	if err := zclient.Start(); err != nil {
		log.S().Fatalf("failed to start zone client: %v", err)
	}
	var realms []*proto.Realm
	var entries = make(map[string]*proto.DirEntry)
	if resp, err := zclient.Client().Locate(callCtx, &proto.LocateReq{
		Tenant:     options.Tenant,
		MountPoint: options.Volume,
		Create:     true,
		Merged:     options.Merged,
		Policy: &proto.StoragePolicy{
			StackedTypes: []*proto.StorageType{
				{
					Type: proto.StorageType_SSD,
					Properties: []proto.StorageType_MediumProperty{
						proto.StorageType_RANDOM_ACCESS,
					},
				},
				{
					Type: proto.StorageType_HDD,
					Properties: []proto.StorageType_MediumProperty{
						proto.StorageType_RANDOM_ACCESS,
					},
				},
				{
					Type: proto.StorageType_BLOCK_STORAGE,
					Properties: []proto.StorageType_MediumProperty{
						proto.StorageType_RANDOM_ACCESS,
					},
				},
			},
		},
	}); err != nil {
		log.S().Fatalf("failed to get realm: %v", err)
	} else if len(resp.Realms) == 0 {
		log.S().Fatalf("failed to get zone %v realms", z.Name)
	} else {
		log.S().Infof("get realms %v", resp.Realms)
		realms = resp.Realms
	}
	var clients = make([]*rclient.Client, len(realms))
	var token string
	for i := range realms {
		rclient := rclient.NewClient(realms[i].Address)
		if err := rclient.Start(); err != nil {
			log.S().Fatalf("failed to start realm client: %v", err)
		}
		if resp, err := rclient.Client().Mount(callCtx, &proto.MountReq{
			Tenant: options.Tenant,
			Name:   options.Volume,
			Merged: options.Merged,
		}); err != nil {
			log.S().Fatalf("failed to mount in realm: %v", err)
		} else if len(resp.MountPoint.Realms) == 0 {
			log.S().Fatalf("failed to mount in realm %s", z.Name)
		} else if len(resp.Token) == 0 {
			log.S().Fatalf("failed to get token in realm %s", z.Name)
		} else if len(token) > 0 && token != resp.Token {
			log.S().Fatalf("get wrong token %s in realm %s", resp.Token, z.Name)
		} else {
			token = resp.Token
			log.S().Infof("mount %s in realm %s address %s token %s", options.Volume, realms[i].Name, realms[i].Address, resp.Token)
			for i := range resp.Entries {
				entry := resp.Entries[i]
				if ent, ok := entries[entry.Name]; ok {
					entry = mergefs.CompareEntry(ent, entry)
				}
				entries[entry.Name] = entry
			}
		}
		clients[i] = rclient
	}

	opts := &fs.Options{
		AttrTimeout:  &cacheDuration,
		EntryTimeout: &cacheDuration,
		MountOptions: fuse.MountOptions{
			FsName:     "ffs",
			MaxWrite:   MaxKernelWrite,
			AllowOther: true,
		},
		UID: uid,
		GID: gid,
	}

	ctx := &Context{
		uclient:    uclient,
		zclient:    zclient,
		client:     clients[0],
		clients:    clients,
		realms:     realms,
		zone:       options.Zone,
		tenant:     options.Tenant,
		volume:     options.Volume,
		target:     options.Target,
		token:      token,
		merged:     options.Merged,
		asyncWrite: options.AsyncWrite,
		cacheDir:   options.CacheDir,
		keepCache:  options.KeepCache,
		wg:         &sync.WaitGroup{},
	}
	ctx.mergedfs.entries = entries

	ctx.ctx, ctx.cancel = context.WithCancel(context.Background())
	for i := range ctx.clients {
		idx := i
		client := ctx.clients[idx]
		ctx.wg.Add(1)
		go func() {
			defer ctx.wg.Done()
			ticker := time.NewTicker(time.Second)
			for {
				callCtx, cancel := context.WithTimeout(context.Background(), time.Second)
				if _, err := client.Client().Heartbeat(callCtx, &proto.MountPointHeartbeatReq{
					Token: ctx.token,
				}); err != nil {
					if fserror.IsNotMount(err) {
						if resp, err := client.Client().Mount(callCtx, &proto.MountReq{
							Tenant: ctx.tenant,
							Name:   ctx.volume,
							Merged: ctx.merged,
						}); err != nil {
							log.S().Errorf("failed to mount in realm: %v", err)
						} else if len(resp.MountPoint.Realms) == 0 {
							log.S().Errorf("failed to mount in realm %s", z.Name)
						} else if len(resp.Token) == 0 {
							log.S().Errorf("failed to get token in realm %s", z.Name)
						} else if len(ctx.token) > 0 && ctx.token != resp.Token {
							log.S().Errorf("get wrong token %s in realm %s", resp.Token, z.Name)
						} else {
							log.S().Infof("mount %s in realm %s address %s token %s", options.Volume, realms[i].Name, realms[i].Address, resp.Token)
						}
					} else if ctx.merged && ctx.clientIndex() == uint64(idx) {
						ctx.Next()
					}
					if ctx.merged {
						log.S().Warnf("failed to heartbeat: %v", err)
					} else {
						log.S().Errorf("failed to heartbeat: %v", err)
					}
				}
				cancel()
				select {
				case <-ticker.C:
				case <-ctx.ctx.Done():
					ticker.Stop()
					return
				}
			}
		}()
	}
	ctx.wg.Add(1)
	go func() {
		defer ctx.wg.Done()
		for {
			stream, err := ctx.uclient.Client().WatchZones(ctx.ctx, &proto.WatchZonesReq{Pattern: z.Name})
			if err != nil {
				log.S().Errorf("failed to watch zones: %v", err)
			} else {
				for {
					zonesList, err := stream.Recv()
					if err != nil {
						log.S().Warn(err)
						break
					}
					for _, zone := range zonesList.Zones {
						if zone.Name == z.Name {
							ctx.zclient.Update(zone.Endpoints...)
							log.S().Warnf("update zone %s endpoints %v", zone.Name, zone.Endpoints)
							break
						}
					}
				}
			}
			timer := time.NewTimer(time.Second)
			select {
			case <-timer.C:
			case <-ctx.ctx.Done():
				timer.Stop()
				return
			}
		}
	}()
	for i := range realms {
		idx := i
		r := realms[idx]
		ctx.wg.Add(1)
		go func() {
			defer ctx.wg.Done()
			for {
				stream, err := ctx.zclient.Client().WatchRealms(ctx.ctx, &proto.WatchRealmsReq{Pattern: r.Name})
				if err != nil {
					log.S().Errorf("failed to watch realms: %v", err)
				} else {
					log.S().Infof("watch realm %s", r.Name)
					for {
						realmsList, err := stream.Recv()
						if err != nil {
							log.S().Warn(err)
							break
						}
						for _, realm := range realmsList.Realms {
							if realm.Name == r.Name {
								ctx.realms[idx] = realm
								ctx.clients[idx].Update(realm.Address)
								log.S().Warnf("update realm %s address %s", realm.Name, realm.Address)
								break
							}
						}
					}
				}
				timer := time.NewTimer(time.Second)
				select {
				case <-timer.C:
				case <-ctx.ctx.Done():
					timer.Stop()
					return
				}
			}
		}()
	}

	if len(options.CacheDir) > 0 {
		if options.CacheCap <= 0 {
			options.CacheCap = 1024 * 1024 * 1024
		}
		ctx.cache.LRU = lru.New(options.CacheCap, func(key, value interface{}) {
			if rn, ok := value.(*RfsNode); ok {
				if cacheFile := rn.cacheFile; cacheFile != nil && cacheFile.immutable {
					rn.removeCacheFile()
				}
			}
		})
	}

	log.S().Infof("mounting %s", options.Target)
	server, err := fs.Mount(options.Target, NewRootFS(ctx), opts)
	if err != nil {
		log.S().Fatalf("failed to mount, %v", err)
		return nil
	}
	log.S().Info("mount success")
	return &Server{
		Server: server,
		ctx:    ctx,
	}
}

func (s *Server) ListenForUnmount() {
	c := make(chan os.Signal, 1)
	signal.Notify(c, syscall.SIGHUP,
		syscall.SIGINT,
		syscall.SIGTERM,
		syscall.SIGQUIT)
	sig := <-c
	log.S().Infof("Got %s signal, unmounting %q...", sig, s.ctx.target)
	s.Stop()
	err := s.Unmount()
	if err != nil {
		log.S().Errorf("Failed to unmount, try %q manually.", "umount "+s.ctx.target)
	}
	<-c // Double ctrl+c
	log.S().Warn("Force exiting...")
	os.Exit(1)
}

func (s *Server) Stop() {
	callCtx, cancel := context.WithTimeout(context.Background(), time.Second*3)
	defer cancel()
	for i := range s.ctx.clients {
		client := s.ctx.clients[i]
		if _, err := client.Client().Umount(callCtx, &proto.UmountReq{
			Tenant: s.ctx.tenant,
			Name:   s.ctx.volume,
		}); err != nil {
			log.S().Errorf("failed to umount in realm %s address %s err %v", s.ctx.realms[i].Name, s.ctx.realms[i].Address, err)
		} else {
			log.S().Infof("umount %s in realm %s address %s", s.ctx.volume, s.ctx.realms[i].Name, s.ctx.realms[i].Address)
		}
	}
	s.ctx.cancel()
	s.ctx.wg.Wait()
	s.ctx.uclient.Stop()
	s.ctx.zclient.Stop()
}