package realm

import (
	"context"
	"crypto/sha256"
	"encoding/base64"
	"io"
	_ "net/http/pprof"
	"os"
	"path"
	"path/filepath"
	"strings"
	"sync"
	"syscall"
	"time"

	"github.com/pingcap/log"
	"github.com/shirou/gopsutil/v3/disk"
	"github.com/tidbcloud/ffs/pkg/v1/proto"
	"github.com/tidbcloud/ffs/realm/fserror"
	"github.com/tidbcloud/ffs/realm/mergefs"
	"github.com/tidbcloud/ffs/realm/metrics"
	uclient "github.com/tidbcloud/ffs/universe/client"
	zclient "github.com/tidbcloud/ffs/zone/client"
)

const maxRetryTimes = 5

type RealmServer struct {
	proto.UnimplementedRealmServiceServer
	proto.UnimplementedRealmControlServiceServer
	ctx     context.Context
	cancel  context.CancelFunc
	uclient *uclient.Client
	zclient *zclient.Client
	name    string
	address string
	rootDir string
	gitHash string
	mounts  struct {
		sync.RWMutex
		metas map[string]*mountMeta
	}
}

type mountMeta struct {
	*proto.MountReq
	token string
	mu    struct {
		sync.RWMutex
		files map[fileHandleKey]*fileHandle
	}
}

func newFileHandleKey(fh *proto.FileHandle) fileHandleKey {
	return fileHandleKey{
		name:  fh.GetName(),
		flags: fh.GetFlags(),
	}
}

type fileHandleKey struct {
	name  string
	flags uint32
}

type fileHandle struct {
	*os.File
	appendMode bool
	segFile    mergefs.SegFile
}

func NewRealmServer(universe, zone, name, address, root string, gitHash string) *RealmServer {
	callCtx, cancel := context.WithTimeout(context.Background(), time.Second*15)
	defer cancel()
	uclient := uclient.NewClient(strings.Split(universe, ",")...)
	if err := uclient.Start(); err != nil {
		log.S().Fatalf("failed to start client: %v", err)
	}
	var err error
	var z *proto.Zone
	for i := 0; i < maxRetryTimes; i++ {
		var resp *proto.Zone
		if resp, err = uclient.Client().GetZone(callCtx, &proto.GetZoneReq{Name: zone}); err != nil {
			log.S().Warnf("failed to get zone: %v", err)
			time.Sleep(time.Second * 3)
			continue
		} else if len(resp.Endpoints) == 0 {
			log.S().Warnf("failed to get zone %v endpoints", resp.Name)
			time.Sleep(time.Second * 3)
			continue
		} else {
			log.S().Infof("get zone name: %s endpoints: %s", resp.Name, resp.Endpoints)
			z = resp
		}
	}
	if err != nil {
		log.S().Fatalf("failed to get zone: %v", err)
	} else if z == nil {
		log.S().Fatalf("failed to get zone %v endpoints", zone)
	}
	log.S().Info(z.Endpoints)
	zclient := zclient.NewClient(z.Endpoints...)
	if err := zclient.Start(); err != nil {
		log.S().Fatalf("failed to start zone client: %v", err)
	}
	svr := &RealmServer{
		name:    name,
		address: address,
		gitHash: gitHash,
		uclient: uclient,
		zclient: zclient,
	}
	svr.mounts.metas = make(map[string]*mountMeta)
	svr.rootDir = root
	svr.ctx, svr.cancel = context.WithCancel(context.Background())
	go func() {
		for {
			stream, err := svr.uclient.Client().WatchZones(svr.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 {
							svr.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 <-svr.ctx.Done():
				timer.Stop()
				return
			}
		}
	}()
	go func() {
		ticker := time.NewTicker(time.Second)
		policy := &proto.StoragePolicy{
			StackedTypes: []*proto.StorageType{
				{
					Type: proto.StorageType_SSD,
					Properties: []proto.StorageType_MediumProperty{
						proto.StorageType_RANDOM_ACCESS,
					},
				},
			},
		}

		for {
			diskStat, err := disk.Usage(svr.rootDir)
			if err != nil {
				log.S().Errorf("failed to get disk usage err %s", err)
				time.Sleep(time.Second)
				continue
			}
			req := &proto.RealmHeartbeatReq{
				Realm: &proto.Realm{
					Name:    name,
					Address: address,
					RuntimeInfo: &proto.RealmRuntimeInfo{
						Capacity:        int64(diskStat.Total),
						Used:            int64(diskStat.Used),
						LastUpdate:      time.Now().UnixNano(),
						SoftwareVersion: svr.gitHash,
					},
					Policy: policy,
				},
			}
			if _, err := svr.zclient.Client().RealmHeartbeat(svr.ctx, req); err != nil {
				log.S().Errorf("failed to heartbeat: %v", err)
			}
			select {
			case <-ticker.C:
			case <-svr.ctx.Done():
				ticker.Stop()
				return
			}
		}
	}()
	return svr
}

func (s *RealmServer) toFullPath(token, name string) string {
	return path.Join(s.rootDir, token, name)
}

func (s *RealmServer) mountMeta(token string) *mountMeta {
	s.mounts.RLock()
	m := s.mounts.metas[token]
	s.mounts.RUnlock()
	return m
}

func (s *RealmServer) Stop() {
	s.uclient.Stop()
	s.zclient.Stop()
	s.cancel()
}

func mountPath(mountPath, name string) string {
	h := sha256.New()
	h.Write([]byte(mountPath))
	h.Write([]byte(name))
	return base64.RawURLEncoding.EncodeToString(h.Sum(nil))
}

func (s *RealmServer) Mount(ctx context.Context, req *proto.MountReq) (*proto.MountResp, error) {
	if len(req.Tenant) == 0 {
		return nil, fserror.ErrNotMount
	}
	log.S().Infof("mount tenant %s name %s", req.Tenant, req.Name)
	token := mountPath(req.Tenant, req.Name)

	fullPath := s.toFullPath(token, "")
	s.mounts.Lock()
	m, ok := s.mounts.metas[token]
	if !ok {
		m = &mountMeta{MountReq: req, token: token}
		m.mu.files = make(map[fileHandleKey]*fileHandle)
		s.mounts.metas[token] = m
		s.mounts.Unlock()
		if err := makeDir(fullPath); err != nil {
			log.S().Errorf("failed to create mount dir: %v", err)
			return nil, fserror.ErrNotMount
		}
	} else {
		s.mounts.Unlock()
	}
	var entries []*proto.DirEntry
	if m.Merged {
		entries = mergefs.Entries(fullPath)
	}
	return &proto.MountResp{
		MountPoint: &proto.MountPoint{
			Realms: []*proto.Realm{{Name: s.name, Address: s.address}},
		},
		Token:   m.token,
		Entries: entries,
	}, nil
}

func (s *RealmServer) Umount(ctx context.Context, req *proto.UmountReq) (*proto.EmptyMessage, error) {
	if len(req.Tenant) == 0 {
		return nil, fserror.ErrNotMount
	}
	token := mountPath(req.Tenant, req.Name)
	log.S().Infof("umount tenant %s name %s", req.Tenant, req.Name)
	s.mounts.Lock()
	if m, ok := s.mounts.metas[token]; ok {
		m.MountReq = &proto.MountReq{}
		delete(s.mounts.metas, token)
		s.mounts.Unlock()
		m.mu.Lock()
		for _, fh := range m.mu.files {
			fh.Close()
		}
		m.mu.files = map[fileHandleKey]*fileHandle{}
		m.mu.Unlock()
	} else {
		s.mounts.Unlock()
	}
	return &proto.EmptyMessage{}, nil
}

func (s *RealmServer) Cleanup(ctx context.Context, req *proto.CleanupReq) (*proto.EmptyMessage, error) {
	if len(req.Tenant) == 0 {
		return nil, fserror.ErrNotMount
	}
	token := mountPath(req.Tenant, req.Name)
	fullPath := s.toFullPath(token, "")
	log.S().Infof("cleanup tenant %s name %s", req.Tenant, req.Name)
	s.mounts.Lock()
	if m, ok := s.mounts.metas[token]; ok {
		m.MountReq = &proto.MountReq{}
		delete(s.mounts.metas, token)
		s.mounts.Unlock()
	} else {
		s.mounts.Unlock()
	}
	err := os.RemoveAll(fullPath)
	if err != nil {
		log.S().Errorf("failed to remove %s err %v", fullPath, err)
		return nil, err
	}
	return &proto.EmptyMessage{}, nil
}

func (s *RealmServer) Heartbeat(ctx context.Context, req *proto.MountPointHeartbeatReq) (*proto.MountPointHeartbeatResp, error) {
	//log.S().Debugf("heartbeat")
	m := s.mountMeta(req.GetToken())
	if m == nil {
		return nil, fserror.ErrNotMount
	}
	return &proto.MountPointHeartbeatResp{}, nil
}

func (s *RealmServer) Create(ctx context.Context, req *proto.CreateReq) (*proto.CreateResp, error) {
	m := s.mountMeta(req.GetFileHandle().GetToken())
	if m == nil {
		return nil, fserror.ErrNotMount
	}
	fullPath := s.toFullPath(m.token, req.FileHandle.Name)
	if m.Merged {
		dir := filepath.Dir(fullPath)
		_, err := os.Stat(dir)
		if err != nil && os.IsNotExist(err) {
			os.MkdirAll(dir, os.ModePerm)
		}
		f, err := mergefs.OpenSegFile(fullPath)
		if err != nil {
			log.S().Warn("Create %s failed, err:%v", fullPath, err)
			return nil, err
		}

		key := newFileHandleKey(req.FileHandle)
		log.S().Infof("create merged %s, flags %v file handle %v", fullPath, req.FileHandle.Flags, key)

		h := &fileHandle{segFile: f}

		m.mu.Lock()
		m.mu.files[key] = h
		m.mu.Unlock()

		fi, err := f.Stat()
		if err != nil {
			log.S().Warnf("stat %s failed: %v", fullPath, err)
			return nil, err
		}
		e := &proto.DirEntry{
			IsDir:   fi.IsDir(),
			Mode:    uint32(fi.Mode()),
			Sz:      uint64(fi.Size()),
			ModTime: uint64(fi.ModTime().UnixNano()),
		}
		if s, ok := fi.Sys().(*syscall.Stat_t); ok {
			e.Nlink = uint32(s.Nlink)
			e.Ino = uint64(s.Ino)
		}

		return &proto.CreateResp{Entry: e}, nil
	}
	f, err := os.OpenFile(fullPath, int(req.FileHandle.Flags), os.FileMode(req.Mode))
	if err != nil {
		log.S().Warn("Create %s failed, err:%v", fullPath, err)
		return nil, err
	}

	key := newFileHandleKey(req.FileHandle)
	log.S().Infof("create %s, flags %v file handle %v", fullPath, req.FileHandle.Flags, key)

	h := &fileHandle{File: f, appendMode: int(req.FileHandle.Flags)&syscall.O_APPEND != 0}

	m.mu.Lock()
	m.mu.files[key] = h
	m.mu.Unlock()

	fi, err := f.Stat()
	if err != nil {
		log.S().Warnf("stat %s failed: %v", fullPath, err)
		return nil, err
	}
	e := &proto.DirEntry{
		IsDir:   fi.IsDir(),
		Mode:    uint32(fi.Mode()),
		Sz:      uint64(fi.Size()),
		ModTime: uint64(fi.ModTime().UnixNano()),
	}
	if s, ok := fi.Sys().(*syscall.Stat_t); ok {
		e.Nlink = uint32(s.Nlink)
		e.Ino = uint64(s.Ino)
	}

	return &proto.CreateResp{Entry: e}, nil
}

func (s *RealmServer) Open(ctx context.Context, req *proto.OpenReq) (*proto.OpenResp, error) {
	startTime := time.Now()
	m := s.mountMeta(req.GetFileHandle().GetToken())
	if m == nil {
		return nil, fserror.ErrNotMount
	}
	fullPath := s.toFullPath(m.token, req.FileHandle.Name)
	key := newFileHandleKey(req.FileHandle)

	if m.Merged {
		f, err := mergefs.OpenSegFile(fullPath)
		if err != nil {
			if os.IsNotExist(err) {
				log.S().Warnf("Open %s failed, err:%v", fullPath, err)
				return &proto.OpenResp{Errno: int64(syscall.ENOENT)}, nil
			}
			log.S().Errorf("Open %s failed, err:%v", fullPath, err)
			return nil, err
		}

		log.S().Infof("open merged %s, flags %v, file handle %v", fullPath, req.FileHandle.Flags, key)

		h := &fileHandle{segFile: f}

		m.mu.Lock()
		old, ok := m.mu.files[key]
		m.mu.files[key] = h
		m.mu.Unlock()
		if ok {
			old.Close()
		}
		metrics.OpenWaitTimeDurationHistogram.Observe(time.Since(startTime).Seconds())

		return &proto.OpenResp{}, nil
	}
	fi, err := os.Stat(fullPath)
	if err != nil {
		if os.IsNotExist(err) {
			log.S().Warnf("Stat %s failed, err:%v", fullPath, err)
			return &proto.OpenResp{Errno: int64(syscall.ENOENT)}, nil
		}
		return nil, err
	}
	flags := int(req.FileHandle.Flags)
	if !fi.IsDir() {
		// After the file is created, the realm restarts and opens the file again. The flags need to filter O_CREATE.
		if flags&os.O_CREATE > 0 {
			flags = flags ^ os.O_CREATE
		}
	}
	f, err := os.OpenFile(fullPath, flags, 0666)
	if err != nil {
		if os.IsNotExist(err) {
			log.S().Warnf("Open %s failed, err:%v", fullPath, err)
			return &proto.OpenResp{Errno: int64(syscall.ENOENT)}, nil
		}
		log.S().Errorf("Open %s failed, err:%v", fullPath, err)
		return nil, err
	}

	log.S().Infof("open %s, flags %v, file handle %v", fullPath, req.FileHandle.Flags, key)

	h := &fileHandle{File: f, appendMode: int(req.FileHandle.Flags)&syscall.O_APPEND != 0}

	m.mu.Lock()
	old, ok := m.mu.files[key]
	m.mu.files[key] = h
	m.mu.Unlock()
	if ok {
		old.Close()
	}

	metrics.OpenWaitTimeDurationHistogram.Observe(time.Since(startTime).Seconds())

	return &proto.OpenResp{}, nil
}

func (s *RealmServer) Readdir(ctx context.Context, req *proto.ReaddirReq) (*proto.ReaddirResp, error) {
	m := s.mountMeta(req.GetToken())
	if m == nil {
		return nil, fserror.ErrNotMount
	}
	fullPath := s.toFullPath(m.token, req.Name)
	f, err := os.Open(fullPath)
	if err != nil {
		log.S().Warnf("open %s failed: %v", fullPath, err)
		return nil, err
	}
	if err != nil {
		if os.IsNotExist(err) {
			return &proto.ReaddirResp{Errno: int64(syscall.ENOENT)}, nil
		}
		log.S().Warnf("stat %s failed: %v", fullPath, err)
		return nil, err
	}

	names, err := f.Readdirnames(-1)
	if err != nil {
		log.S().Warnf("readdir %s failed:%v", fullPath, err)
		return nil, err
	}
	log.S().Debugf("readdir %s, lists:%v", fullPath, names)
	entries := make([]*proto.DirEntry, 0, len(names))
	for _, n := range names {
		// stat each file
		fullPath := path.Join(fullPath, n)

		fi, err := os.Lstat(fullPath)
		if err != nil {
			log.S().Warnf("stat %s failed: %v", fullPath, err)
			return nil, err
		}
		e := &proto.DirEntry{
			IsDir:   fi.IsDir(),
			Name:    n,
			Mode:    uint32(fi.Mode()),
			Sz:      uint64(fi.Size()),
			ModTime: uint64(fi.ModTime().UnixNano()),
		}
		if s, ok := fi.Sys().(*syscall.Stat_t); ok {
			e.Nlink = uint32(s.Nlink)
			e.Ino = uint64(s.Ino)
		}

		if !fi.IsDir() && m.Merged {
			f, err := mergefs.OpenSegFile(fullPath)
			if err != nil {
				if !fserror.IsDir(err) {
					log.S().Warnf("open seg file %s failed: %v", fullPath, err)
					return nil, err
				}
			} else {
				defer f.Close()
				fi, err = f.Stat()
				if err != nil {
					log.S().Warnf("stat %s failed: %v", fullPath, err)
					return nil, err
				}
				e.Sz = uint64(fi.Size())
			}
		}
		entries = append(entries, e)
	}
	return &proto.ReaddirResp{Entries: entries}, nil
}

func (s *RealmServer) Mknod(ctx context.Context, req *proto.MknodReq) (*proto.MknodResp, error) {
	m := s.mountMeta(req.GetToken())
	if m == nil {
		return nil, fserror.ErrNotMount
	}
	fullPath := s.toFullPath(m.token, req.Name)
	err := syscall.Mknod(fullPath, uint32(req.Mode), int(req.Dev))
	if err != nil {
		log.S().Warnf("mknod %s failed: %v", fullPath, err)
		return nil, err
	}
	log.S().Debugf("mknod %s, mode %d, dev %d", fullPath, req.Mode, req.Dev)

	fi, err := os.Stat(fullPath)
	if err != nil {
		log.S().Warnf("stat %s failed: %v", fullPath, err)
		return nil, err
	}
	e := &proto.DirEntry{
		IsDir:   fi.IsDir(),
		Mode:    uint32(fi.Mode()),
		Sz:      uint64(fi.Size()),
		ModTime: uint64(fi.ModTime().UnixNano()),
	}

	if !fi.IsDir() && m.Merged {
		f, err := mergefs.OpenSegFile(fullPath)
		if err != nil {
			log.S().Warnf("open seg file %s failed: %v", fullPath, err)
			return nil, err
		}
		defer f.Close()
		fi, err = f.Stat()
		if err != nil {
			log.S().Warnf("stat %s failed: %v", fullPath, err)
			return nil, err
		}
		e.Sz = uint64(fi.Size())
	}

	if s, ok := fi.Sys().(*syscall.Stat_t); ok {
		e.Nlink = uint32(s.Nlink)
		e.Ino = uint64(s.Ino)
	}
	return &proto.MknodResp{Entry: e}, nil
}

func (s *RealmServer) Mkdir(ctx context.Context, req *proto.MkdirReq) (*proto.MkdirResp, error) {
	m := s.mountMeta(req.GetToken())
	if m == nil {
		return nil, fserror.ErrNotMount
	}
	fullPath := s.toFullPath(m.token, req.Name)
	err := os.MkdirAll(fullPath, os.FileMode(req.Mode))
	if err != nil {
		log.S().Warnf("mkdir %s failed: %v", fullPath, err)
		return nil, err
	}

	fi, err := os.Stat(fullPath)
	if err != nil {
		log.S().Warnf("stat %s failed: %v", fullPath, err)
		return nil, err
	}
	e := &proto.DirEntry{
		IsDir:   fi.IsDir(),
		Mode:    uint32(fi.Mode()),
		Sz:      uint64(fi.Size()),
		ModTime: uint64(fi.ModTime().UnixNano()),
	}

	if !fi.IsDir() && m.Merged {
		f, err := mergefs.OpenSegFile(fullPath)
		if err != nil {
			log.S().Warnf("open seg file %s failed: %v", fullPath, err)
			return nil, err
		}
		defer f.Close()
		fi, err = f.Stat()
		if err != nil {
			log.S().Warnf("stat %s failed: %v", fullPath, err)
			return nil, err
		}
		e.Sz = uint64(fi.Size())
	}

	if s, ok := fi.Sys().(*syscall.Stat_t); ok {
		e.Nlink = uint32(s.Nlink)
		e.Ino = uint64(s.Ino)
	}
	return &proto.MkdirResp{Entry: e}, nil
}

func (s *RealmServer) Rmdir(ctx context.Context, req *proto.RmdirReq) (*proto.RmdirResp, error) {
	m := s.mountMeta(req.GetToken())
	if m == nil {
		return nil, fserror.ErrNotMount
	}
	fullPath := s.toFullPath(m.token, req.Name)
	err := syscall.Rmdir(fullPath)
	if err != nil {
		if err == syscall.ENOTEMPTY {
			return &proto.RmdirResp{Errno: int64(syscall.ENOTEMPTY)}, nil
		}
		log.S().Warnf("rmdir %s failed: %v", fullPath, err)
		return nil, err
	}
	return &proto.RmdirResp{}, nil
}

func (s *RealmServer) Unlink(ctx context.Context, req *proto.UnlinkReq) (*proto.UnlinkResp, error) {
	m := s.mountMeta(req.GetToken())
	if m == nil {
		return nil, fserror.ErrNotMount
	}
	fullPath := s.toFullPath(m.token, req.Name)
	log.S().Infof("unlink %s", fullPath)
	err := syscall.Unlink(fullPath)
	if err != nil {
		if !m.Merged {
			log.S().Error(err)
		}
		return nil, err
	}
	return &proto.UnlinkResp{}, nil
}

func (s *RealmServer) Lookup(ctx context.Context, req *proto.LookupReq) (*proto.LookupResp, error) {
	m := s.mountMeta(req.GetToken())
	if m == nil {
		return nil, fserror.ErrNotMount
	}
	fullPath := s.toFullPath(m.token, req.Name)
	log.S().Debugf("lookup %s", fullPath)

	fi, err := os.Lstat(fullPath)
	if err != nil {
		if os.IsNotExist(err) {
			return &proto.LookupResp{Errno: int64(syscall.ENOENT)}, nil
		}
		log.S().Warnf("stat %s failed: %v", fullPath, err)
		return nil, err
	}
	e := &proto.DirEntry{
		IsDir:   fi.IsDir(),
		Mode:    uint32(fi.Mode()),
		Sz:      uint64(fi.Size()),
		ModTime: uint64(fi.ModTime().UnixNano()),
	}
	if s, ok := fi.Sys().(*syscall.Stat_t); ok {
		e.Nlink = uint32(s.Nlink)
		e.Ino = uint64(s.Ino)
	}

	if !fi.IsDir() && m.Merged {
		f, err := mergefs.OpenSegFile(fullPath)
		if err != nil {
			log.S().Warnf("open seg file %s failed: %v", fullPath, err)
			return nil, err
		}
		defer f.Close()
		fi, err = f.Stat()
		if err != nil {
			log.S().Warnf("stat %s failed: %v", fullPath, err)
			return nil, err
		}
		e.Sz = uint64(fi.Size())
	}
	log.S().Debugf("lookup %s entry %v", fullPath, e)
	return &proto.LookupResp{Entry: e}, nil
}

func (s *RealmServer) Link(ctx context.Context, req *proto.LinkReq) (*proto.LinkResp, error) {
	m := s.mountMeta(req.GetToken())
	if m == nil {
		return nil, fserror.ErrNotMount
	}
	oldFullPath := s.toFullPath(m.token, req.OldName)
	newFullPath := s.toFullPath(m.token, req.NewName)
	log.S().Infof("link %s -> %s", oldFullPath, newFullPath)

	err := syscall.Link(oldFullPath, newFullPath)
	if err != nil {
		log.S().Error(err)
		return nil, err
	}

	return &proto.LinkResp{}, nil
}

func (s *RealmServer) Symlink(ctx context.Context, req *proto.SymlinkReq) (*proto.SymlinkResp, error) {
	m := s.mountMeta(req.GetToken())
	if m == nil {
		return nil, fserror.ErrNotMount
	}
	newFullPath := s.toFullPath(m.token, req.NewName)
	log.S().Infof("symlink %s -> %s", req.OldName, newFullPath)

	err := syscall.Symlink(req.OldName, newFullPath)
	if err != nil {
		log.S().Warn(err)
		switch t := err.(type) {
		case syscall.Errno:
			return &proto.SymlinkResp{Errno: int64(t)}, nil
		case *os.SyscallError:
			return &proto.SymlinkResp{Errno: int64(t.Err.(syscall.Errno))}, nil
		}
		return nil, err
	}
	fi, err := os.Lstat(newFullPath)
	if err != nil {
		if os.IsNotExist(err) {
			return &proto.SymlinkResp{Errno: int64(syscall.ENOENT)}, nil
		}
		log.S().Warnf("stat %s failed: %v", newFullPath, err)
		return nil, err
	}
	e := &proto.DirEntry{
		IsDir:   fi.IsDir(),
		Mode:    uint32(fi.Mode()),
		Sz:      uint64(fi.Size()),
		ModTime: uint64(fi.ModTime().UnixNano()),
	}
	if s, ok := fi.Sys().(*syscall.Stat_t); ok {
		e.Nlink = uint32(s.Nlink)
		e.Ino = uint64(s.Ino)
	}
	return &proto.SymlinkResp{Entry: e}, nil
}

func (s *RealmServer) Readlink(ctx context.Context, req *proto.ReadlinkReq) (*proto.ReadlinkResp, error) {
	m := s.mountMeta(req.GetToken())
	if m == nil {
		return nil, fserror.ErrNotMount
	}
	fullPath := s.toFullPath(m.token, req.Name)
	log.S().Infof("readlink %s", fullPath)

	bufSize := 64
	for {
		buf := make([]byte, bufSize)
		n, err := syscall.Readlink(fullPath, buf)
		if err != nil {
			log.S().Warnf("readlink %s failed: %v", fullPath, err)
			switch t := err.(type) {
			case syscall.Errno:
				return &proto.ReadlinkResp{Errno: int64(t)}, nil
			case *os.SyscallError:
				return &proto.ReadlinkResp{Errno: int64(t.Err.(syscall.Errno))}, nil
			}
			return nil, err
		}
		if n < len(buf) {
			return &proto.ReadlinkResp{Buf: buf[:n]}, nil
		}
		bufSize *= 2
	}
}

func (s *RealmServer) Rename(ctx context.Context, req *proto.RenameReq) (*proto.RenameResp, error) {
	m := s.mountMeta(req.GetToken())
	if m == nil {
		return nil, fserror.ErrNotMount
	}
	srcFullPath := s.toFullPath(m.token, req.OldName)
	dstFullPath := s.toFullPath(m.token, req.NewName)
	log.S().Infof("rename %s -> %s", srcFullPath, dstFullPath)

	// rename file
	err := os.Rename(srcFullPath, dstFullPath)
	if err != nil {
		log.S().Error(err)
		return nil, err
	}

	return &proto.RenameResp{}, nil
}

func (s *RealmServer) Statfs(ctx context.Context, req *proto.StatfsReq) (*proto.StatfsResp, error) {
	m := s.mountMeta(req.GetToken())
	if m == nil {
		return nil, fserror.ErrNotMount
	}
	fullPath := s.toFullPath(m.token, req.Name)
	log.S().Infof("statfs %s", fullPath)

	var statfs syscall.Statfs_t
	err := syscall.Statfs(fullPath, &statfs)
	if err != nil {
		log.S().Warnf("statfs %s failed: %v", fullPath, err)
		return nil, err
	}
	return convertToStatfsResp(&statfs), nil
}

func (s *RealmServer) Close(ctx context.Context, req *proto.CloseReq) (*proto.CloseResp, error) {
	log.S().Infof("close file handle %v", req.FileHandle)
	m := s.mountMeta(req.GetFileHandle().GetToken())
	if m == nil {
		return nil, fserror.ErrNotMount
	}
	key := newFileHandleKey(req.FileHandle)

	m.mu.Lock()
	f := m.mu.files[key]
	if f != nil {
		delete(m.mu.files, key)
		m.mu.Unlock()
		if f.segFile != nil {
			f.segFile.Close()
		} else {
			f.Close()
		}
	} else {
		m.mu.Unlock()
	}
	return &proto.CloseResp{}, nil
}

func (s *RealmServer) Read(ctx context.Context, req *proto.ReadReq) (*proto.ReadResp, error) {
	startTime := time.Now()
	log.S().Debugf("read file handle %v, off %d", req.FileHandle, req.Off)

	m := s.mountMeta(req.GetFileHandle().GetToken())
	if m == nil {
		return nil, fserror.ErrNotMount
	}
	key := newFileHandleKey(req.FileHandle)
	m.mu.RLock()
	f := m.mu.files[key]
	m.mu.RUnlock()

	if f == nil {
		s.Open(ctx, &proto.OpenReq{FileHandle: req.FileHandle})
		m.mu.RLock()
		f = m.mu.files[key]
		m.mu.RUnlock()
		if f == nil {
			log.S().Warnf("file handle %v not found", req.FileHandle)
			return &proto.ReadResp{Errno: int64(syscall.ENOENT)}, nil
		}
	}

	if f.segFile != nil {
		frames, err := f.segFile.ReadAt(req.Off, req.Len)
		// check EOF error
		if err != nil && err != io.EOF {
			log.S().Errorf("Read file handle %v, err: %v", req.FileHandle, err)
			return nil, err
		}
		length := uint64(0)
		for _, frame := range frames {
			length += frame.Len
		}
		log.S().Debugf("read merged file handle %v, off %d, len %d", req.FileHandle, req.Off, length)
		metrics.ReadSizeByteHistogram.Observe(float64(length))
		metrics.ReadWaitTimeDurationHistogram.Observe(time.Since(startTime).Seconds())
		return &proto.ReadResp{Frames: frames}, nil
	}

	buf := make([]byte, req.Len)
	n, err := f.ReadAt(buf, int64(req.Off))
	// check EOF error
	var done bool
	if err != nil {
		if err != io.EOF {
			log.S().Errorf("Read file handle %v, err: %v", req.FileHandle, err)
			return nil, err
		}
		done = false
	}

	log.S().Debugf("read file handle %v, off %d, len %d", req.FileHandle, req.Off, n)
	metrics.ReadSizeByteHistogram.Observe(float64(len(buf)))
	metrics.ReadWaitTimeDurationHistogram.Observe(time.Since(startTime).Seconds())
	return &proto.ReadResp{Buf: buf[:n], Done: done}, nil
}

func (s *RealmServer) Write(ctx context.Context, req *proto.WriteReq) (*proto.WriteResp, error) {
	startTime := time.Now()
	m := s.mountMeta(req.GetFileHandle().GetToken())
	if m == nil {
		return nil, fserror.ErrNotMount
	}
	key := newFileHandleKey(req.FileHandle)
	m.mu.RLock()
	f := m.mu.files[key]
	m.mu.RUnlock()
	if f == nil {
		s.Open(ctx, &proto.OpenReq{FileHandle: req.FileHandle})
		m.mu.RLock()
		f = m.mu.files[key]
		m.mu.RUnlock()
		if f == nil {
			log.S().Warnf("file handle %v not found", req.FileHandle)
			return &proto.WriteResp{Errno: int64(syscall.ENOENT)}, nil
		}
	}

	if f.segFile != nil {
		log.S().Debugf("write merged file handle %v offset %d, len: %d, appendMode: %t", req.FileHandle, req.Off, len(req.Buf), f.appendMode)
		n, err := f.segFile.WriteAt(req.Buf, int64(req.Off))
		if err != nil {
			log.S().Errorf("write merged file handle %v failed: %v", req.FileHandle, err)
			return nil, err
		}
		metrics.WriteSizeByteHistogram.Observe(float64(len(req.Buf)))
		metrics.WriteWaitTimeDurationHistogram.Observe(time.Since(startTime).Seconds())
		return &proto.WriteResp{Written: int64(n)}, nil
	}
	log.S().Debugf("write file handle %v offset %d, len: %d, appendMode: %t", req.FileHandle, req.Off, len(req.Buf), f.appendMode)

	var n int
	var err error
	if f.appendMode {
		n, err = f.Write(req.Buf)
	} else {
		n, err = f.WriteAt(req.Buf, int64(req.Off))
	}
	if err != nil {
		log.S().Errorf("write file handle %v failed: %v", req.FileHandle, err)
		return nil, err
	}
	metrics.WriteSizeByteHistogram.Observe(float64(len(req.Buf)))
	metrics.WriteWaitTimeDurationHistogram.Observe(time.Since(startTime).Seconds())
	return &proto.WriteResp{Written: int64(n)}, nil
}

func (s *RealmServer) Fsync(ctx context.Context, req *proto.FsyncReq) (*proto.FsyncResp, error) {
	startTime := time.Now()
	log.S().Infof("fsync file handle %v", req.FileHandle)

	m := s.mountMeta(req.GetFileHandle().GetToken())
	if m == nil {
		return nil, fserror.ErrNotMount
	}
	key := newFileHandleKey(req.FileHandle)
	m.mu.RLock()
	f := m.mu.files[key]
	m.mu.RUnlock()

	if f == nil {
		s.Open(ctx, &proto.OpenReq{FileHandle: req.FileHandle})
		m.mu.RLock()
		f = m.mu.files[key]
		m.mu.RUnlock()
		if f == nil {
			return &proto.FsyncResp{Errno: int64(syscall.ENOENT)}, nil
		}
	}
	if f.segFile != nil {
		err := f.segFile.Sync()
		if err != nil {
			log.S().Errorf("fsync merged file handle %v failed: %v", req.FileHandle, err)
			return nil, err
		}

		fi, err := f.segFile.Stat()
		if err != nil {
			if os.IsNotExist(err) {
				return &proto.FsyncResp{Errno: int64(syscall.ENOENT)}, nil
			}
			log.S().Warnf("stat merged file handle %v failed: %v", req.FileHandle, err)
			return nil, err
		}
		metrics.FsyncWaitTimeDurationHistogram.Observe(time.Since(startTime).Seconds())
		return &proto.FsyncResp{Entry: &proto.DirEntry{Sz: uint64(fi.Size()), ModTime: uint64(fi.ModTime().UnixNano())}}, nil
	}
	err := f.Sync()
	if err != nil {
		log.S().Errorf("fsync file handle %v failed: %v", req.FileHandle, err)
		return nil, err
	}

	fi, err := f.Stat()
	if err != nil {
		if os.IsNotExist(err) {
			return &proto.FsyncResp{Errno: int64(syscall.ENOENT)}, nil
		}
		log.S().Warnf("stat file handle %v failed: %v", req.FileHandle, err)
		return nil, err
	}
	if fi.IsDir() {
		s.Close(ctx, &proto.CloseReq{FileHandle: req.FileHandle})
	}
	metrics.FsyncWaitTimeDurationHistogram.Observe(time.Since(startTime).Seconds())
	return &proto.FsyncResp{Entry: &proto.DirEntry{Sz: uint64(fi.Size()), ModTime: uint64(fi.ModTime().UnixNano())}}, nil
}

func (s *RealmServer) Flush(ctx context.Context, req *proto.FlushReq) (*proto.FlushResp, error) {
	startTime := time.Now()
	log.S().Infof("flush file handle %v", req.FileHandle)

	m := s.mountMeta(req.GetFileHandle().GetToken())
	if m == nil {
		return nil, fserror.ErrNotMount
	}
	key := newFileHandleKey(req.FileHandle)
	m.mu.RLock()
	f := m.mu.files[key]
	m.mu.RUnlock()

	if f == nil {
		s.Open(ctx, &proto.OpenReq{FileHandle: req.FileHandle})
		m.mu.RLock()
		f = m.mu.files[key]
		m.mu.RUnlock()
		if f == nil {
			return &proto.FlushResp{Errno: int64(syscall.ENOENT)}, nil
		}
	}
	if f.segFile != nil {
		err := f.segFile.Sync()
		if err != nil {
			log.S().Errorf("fsync merged file handle %v failed: %v", req.FileHandle, err)
			return nil, err
		}

		fi, err := f.segFile.Stat()
		if err != nil {
			if os.IsNotExist(err) {
				return &proto.FlushResp{Errno: int64(syscall.ENOENT)}, nil
			}
			log.S().Warnf("stat merged file handle %v failed: %v", req.FileHandle, err)
			return nil, err
		}
		metrics.FlushWaitTimeDurationHistogram.Observe(time.Since(startTime).Seconds())
		return &proto.FlushResp{Entry: &proto.DirEntry{Sz: uint64(fi.Size()), ModTime: uint64(fi.ModTime().UnixNano())}}, nil
	}
	err := f.Sync()
	if err != nil {
		log.S().Errorf("fsync file handle %v failed: %v", req.FileHandle, err)
		return nil, err
	}

	fi, err := f.Stat()
	if err != nil {
		if os.IsNotExist(err) {
			return &proto.FlushResp{Errno: int64(syscall.ENOENT)}, nil
		}
		log.S().Warnf("stat file handle %v failed: %v", req.FileHandle, err)
		return nil, err
	}
	metrics.FlushWaitTimeDurationHistogram.Observe(time.Since(startTime).Seconds())
	return &proto.FlushResp{Entry: &proto.DirEntry{Sz: uint64(fi.Size()), ModTime: uint64(fi.ModTime().UnixNano())}}, nil
}

func (s *RealmServer) Setattr(ctx context.Context, req *proto.SetattrReq) (*proto.SetattrResp, error) {
	log.S().Infof("Setattr file handle %v, name %s", req.FileHandle, req.Entry.Name)

	if req.FileHandle != nil && len(req.FileHandle.Name) > 0 {

		m := s.mountMeta(req.GetFileHandle().GetToken())
		if m == nil {
			return nil, fserror.ErrNotMount
		}
		key := newFileHandleKey(req.FileHandle)
		m.mu.RLock()
		f := m.mu.files[key]
		m.mu.RUnlock()

		if f == nil {
			s.Open(ctx, &proto.OpenReq{FileHandle: req.FileHandle})
			m.mu.RLock()
			f = m.mu.files[key]
			m.mu.RUnlock()
			if f == nil {
				return &proto.SetattrResp{Errno: int64(syscall.ENOENT)}, nil
			}
		}
		if req.SetMode {
			err := f.Chmod(os.FileMode(req.Entry.Mode))
			if err != nil {
				log.S().Warnf("chmod file handle %v failed: %v", req.FileHandle, err)
				return nil, err
			}
		}

		var e *proto.DirEntry
		if f.segFile != nil {
			if req.SetSize {
				size := req.GetEntry().GetSz()
				err := f.segFile.Truncate(int64(size))
				if err != nil {
					log.S().Warnf("truncate merged file handle %v size %d failed: %v", req.FileHandle, size, err)
					return nil, err
				}
			}

			err := f.segFile.Sync()
			if err != nil {
				log.S().Errorf("fsync merged file handle %v failed: %v", req.FileHandle, err)
				return nil, err
			}

			fi, err := f.segFile.Stat()
			if err != nil {
				if os.IsNotExist(err) {
					return &proto.SetattrResp{Errno: int64(syscall.ENOENT)}, nil
				}
				log.S().Warnf("stat file handle %v failed: %v", req.FileHandle, err)
				return nil, err
			}

			e = &proto.DirEntry{
				IsDir:   fi.IsDir(),
				Mode:    uint32(fi.Mode()),
				Sz:      uint64(fi.Size()),
				ModTime: uint64(fi.ModTime().UnixNano()),
			}
			if s, ok := fi.Sys().(*syscall.Stat_t); ok {
				e.Nlink = uint32(s.Nlink)
				e.Ino = uint64(s.Ino)
			}
		} else {
			if req.SetSize {
				size := req.GetEntry().GetSz()
				err := f.Truncate(int64(size))
				if err != nil {
					log.S().Warnf("truncate file handle %v size %d failed: %v", req.FileHandle, size, err)
					return nil, err
				}
			}
			err := f.Sync()
			if err != nil {
				log.S().Errorf("fsync file handle %v failed: %v", req.FileHandle, err)
				return nil, err
			}

			fi, err := f.Stat()
			if err != nil {
				if os.IsNotExist(err) {
					return &proto.SetattrResp{Errno: int64(syscall.ENOENT)}, nil
				}
				log.S().Warnf("stat file handle %v failed: %v", req.FileHandle, err)
				return nil, err
			}

			e = &proto.DirEntry{
				IsDir:   fi.IsDir(),
				Mode:    uint32(fi.Mode()),
				Sz:      uint64(fi.Size()),
				ModTime: uint64(fi.ModTime().UnixNano()),
			}
			if s, ok := fi.Sys().(*syscall.Stat_t); ok {
				e.Nlink = uint32(s.Nlink)
				e.Ino = uint64(s.Ino)
			}
		}
		return &proto.SetattrResp{Entry: e}, nil
	}
	m := s.mountMeta(req.GetToken())
	if m == nil {
		return nil, fserror.ErrNotMount
	}
	fullPath := s.toFullPath(m.token, req.Entry.Name)
	if req.SetMode {
		err := os.Chmod(fullPath, os.FileMode(req.Entry.Mode))
		if err != nil {
			log.S().Warnf("chmod %v failed: %v", fullPath, err)
			return nil, err
		}
	}

	if req.SetSize {
		size := req.GetEntry().GetSz()
		err := os.Truncate(fullPath, int64(size))
		if err != nil {
			log.S().Warnf("truncate %v size %d failed: %v", fullPath, size, err)
			return nil, err
		}
	}
	fi, err := os.Stat(fullPath)
	if err != nil {
		if os.IsNotExist(err) {
			return &proto.SetattrResp{Errno: int64(syscall.ENOENT)}, nil
		}
		log.S().Warnf("stat %v failed: %v", fullPath, err)
		return nil, err
	}

	e := &proto.DirEntry{
		IsDir:   fi.IsDir(),
		Mode:    uint32(fi.Mode()),
		Sz:      uint64(fi.Size()),
		ModTime: uint64(fi.ModTime().UnixNano()),
	}
	if s, ok := fi.Sys().(*syscall.Stat_t); ok {
		e.Nlink = uint32(s.Nlink)
		e.Ino = uint64(s.Ino)
	}

	if !fi.IsDir() && m.Merged {
		f, err := mergefs.OpenSegFile(fullPath)
		if err != nil {
			log.S().Warnf("open seg file %s failed: %v", fullPath, err)
			return nil, err
		}
		defer f.Close()
		fi, err = f.Stat()
		if err != nil {
			log.S().Warnf("stat %s failed: %v", fullPath, err)
			return nil, err
		}
		e.Sz = uint64(fi.Size())
	}

	return &proto.SetattrResp{Entry: e}, nil
}

func (s *RealmServer) Access(ctx context.Context, req *proto.AccessReq) (*proto.AccessResp, error) {
	m := s.mountMeta(req.GetToken())
	if m == nil {
		return nil, fserror.ErrNotMount
	}
	fullPath := s.toFullPath(m.token, req.Name)
	log.S().Infof("access %s mode %d", fullPath, req.Mode)
	err := syscall.Access(fullPath, req.Mode)
	if err != nil {
		if !m.Merged {
			log.S().Error(err)
		}
		return nil, err
	}
	return &proto.AccessResp{}, nil
}

func (s *RealmServer) Fallocate(ctx context.Context, req *proto.FallocateReq) (*proto.FallocateResp, error) {

	m := s.mountMeta(req.GetFileHandle().GetToken())
	if m == nil {
		return nil, fserror.ErrNotMount
	}
	key := newFileHandleKey(req.FileHandle)
	m.mu.RLock()
	f := m.mu.files[key]
	m.mu.RUnlock()

	if f == nil {
		s.Open(ctx, &proto.OpenReq{FileHandle: req.FileHandle})
		m.mu.RLock()
		f = m.mu.files[key]
		m.mu.RUnlock()
		if f == nil {
			return &proto.FallocateResp{Errno: int64(syscall.ENOENT)}, nil
		}
	}
	log.S().Infof("fallocate %v off %d size %d mode %d", key, req.Off, req.Size_, req.Mode)
	err := fallocate(int(f.Fd()), req.Mode, int64(req.Off), int64(req.Size_))
	if err != nil {
		if !m.Merged {
			log.S().Error(err)
		}
		return nil, err
	}
	return &proto.FallocateResp{}, nil
}

func (s *RealmServer) BatchCommands(stream proto.RealmService_BatchCommandsServer) error {
	h := &batchRequestHandler{
		svr:    s,
		stream: stream,
	}
	return h.start()
}

func (svr *RealmServer) handleBatchRequest(ctx context.Context, req *proto.BatchCommandsRequest_Request) (*proto.BatchCommandsResponse_Response, error) {
	switch req := req.GetCmd().(type) {
	case *proto.BatchCommandsRequest_Request_Read:
		res, err := svr.Read(ctx, req.Read)
		if err != nil {
			return nil, err
		}
		return &proto.BatchCommandsResponse_Response{Cmd: &proto.BatchCommandsResponse_Response_Read{Read: res}}, nil
	case *proto.BatchCommandsRequest_Request_Write:
		res, err := svr.Write(ctx, req.Write)
		if err != nil {
			return nil, err
		}
		return &proto.BatchCommandsResponse_Response{Cmd: &proto.BatchCommandsResponse_Response_Write{Write: res}}, nil
	case *proto.BatchCommandsRequest_Request_Flush:
		res, err := svr.Flush(ctx, req.Flush)
		if err != nil {
			return nil, err
		}
		return &proto.BatchCommandsResponse_Response{Cmd: &proto.BatchCommandsResponse_Response_Flush{Flush: res}}, nil
	case *proto.BatchCommandsRequest_Request_Fsync:
		res, err := svr.Fsync(ctx, req.Fsync)
		if err != nil {
			return nil, err
		}
		return &proto.BatchCommandsResponse_Response{Cmd: &proto.BatchCommandsResponse_Response_Fsync{Fsync: res}}, nil
	case *proto.BatchCommandsRequest_Request_Setattr:
		res, err := svr.Setattr(ctx, req.Setattr)
		if err != nil {
			return nil, err
		}
		return &proto.BatchCommandsResponse_Response{Cmd: &proto.BatchCommandsResponse_Response_Setattr{Setattr: res}}, nil
	case *proto.BatchCommandsRequest_Request_Fallocate:
		res, err := svr.Fallocate(ctx, req.Fallocate)
		if err != nil {
			return nil, err
		}
		return &proto.BatchCommandsResponse_Response{Cmd: &proto.BatchCommandsResponse_Response_Fallocate{Fallocate: res}}, nil
	}
	return nil, nil
}

type batchRequestHandler struct {
	svr    *RealmServer
	stream proto.RealmService_BatchCommandsServer
}

func (h *batchRequestHandler) start() error {
	batchResp := &proto.BatchCommandsResponse{}
	for {
		for i := range batchResp.Responses {
			batchResp.Responses[i] = nil
		}
		batchResp.Responses = batchResp.Responses[:0]
		batchResp.RequestIds = batchResp.RequestIds[:0]
		batchResp.NotMount = false
		var batchReq = proto.BatchCommandsRequest{}
		err := h.stream.RecvMsg(&batchReq)
		if err == io.EOF {
			return nil
		}
		if err != nil {
			return err
		}
		var lastWrite int
		var lastSync int
		var written bool
		var FileHandle *proto.FileHandle
		for i, req := range batchReq.GetRequests() {
			id := batchReq.GetRequestIds()[i]
			switch r := req.GetCmd().(type) {
			case *proto.BatchCommandsRequest_Request_Write:
				written = true
				lastWrite = i
				FileHandle = r.Write.FileHandle
			case *proto.BatchCommandsRequest_Request_Flush, *proto.BatchCommandsRequest_Request_Fsync:
				lastSync = i
			}
			var resp *proto.BatchCommandsResponse_Response
			var err error
			for {
				resp, err = h.svr.handleBatchRequest(context.Background(), req)
				if err != nil {
					if fserror.IsClosed(err) {
						continue
					}
					if fserror.IsNotMount(err) {
						batchResp.NotMount = true
					}
				}
				break
			}
			if batchResp.NotMount {
				break
			}
			if resp != nil {
				batchResp.RequestIds = append(batchResp.RequestIds, id)
				batchResp.Responses = append(batchResp.Responses, resp)
			}
		}
		if written && lastSync <= lastWrite {
			m := h.svr.mountMeta(FileHandle.GetToken())
			if m != nil && m.Merged {
				h.svr.Fsync(context.Background(), &proto.FsyncReq{FileHandle: FileHandle})
			}
		}
		metrics.ServerGrpcReqBatchSize.Observe(float64(len(batchReq.GetRequests())))
		if err := h.stream.SendMsg(batchResp); err != nil {
			return err
		}
		metrics.ServerGrpcRespBatchSize.Observe(float64(len(batchResp.Responses)))
	}
}

func (h *batchRequestHandler) handleRequest(req *proto.BatchCommandsRequest_Request) *proto.BatchCommandsResponse_Response {
	resp, err := h.svr.handleBatchRequest(context.Background(), req)
	if err != nil {
		log.S().Warn("handle batch request failed", err.Error())
		return nil
	}
	return resp
}

func makeDir(pathname string) error {
	err := os.MkdirAll(pathname, os.ModePerm)
	if err != nil {
		if !os.IsExist(err) {
			return err
		}
	}
	return nil
}

func parseDuration(durationStr string) time.Duration {
	dur, err := time.ParseDuration(durationStr)
	if err != nil {
		dur, err = time.ParseDuration(durationStr + "s")
	}
	if err != nil || dur < 0 {
		log.S().Fatalf("invalid duration=%v", durationStr)
	}
	return dur
}