package client

import (
	"context"
	"hash/fnv"
	"os"
	"os/user"
	"path"
	"path/filepath"
	"strconv"
	"strings"
	"sync"
	"syscall"
	"time"
	"unsafe"

	"github.com/hanwen/go-fuse/v2/fs"
	"github.com/hanwen/go-fuse/v2/fuse"
	"github.com/pingcap/log"
	"github.com/tidbcloud/ffs/client/metrics"
	"github.com/tidbcloud/ffs/pkg/v1/proto"
	"github.com/tidbcloud/ffs/realm/mergefs"
)

// The Framework is inspired by etcdfs, thanks to authors

// Set file owners to the current user,
// otherwise in OSX, we will fail to start.
var uid, gid uint32

func init() {
	u, err := user.Current()
	if err != nil {
		panic(err)
	}
	uid32, _ := strconv.ParseUint(u.Uid, 10, 32)
	gid32, _ := strconv.ParseUint(u.Gid, 10, 32)
	uid = uint32(uid32)
	gid = uint32(gid32)
}

type RfsNode struct {
	fs.Inode
	*cacheFile
	ctx      *Context
	e        *proto.DirEntry
	fullPath string
	uid      uint32
	uok      bool
	gid      uint32
	gok      bool
	mu       sync.Mutex
	stream   *batchCommandsClient
	fhs      map[fileHandleKey]int32
}

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

type fileHandleKey struct {
	name  string
	flags uint32
}

type cacheFile struct {
	*os.File
	appendMode bool
	name       string
	wg         sync.WaitGroup
	writes     chan writeOp
	closeCh    chan struct{}
	immutable  bool
	size       int64
	timer      *time.Timer
}

func (f *cacheFile) cached() bool {
	if len(f.name) > 0 && f.size > 0 {
		fi, err := os.Stat(f.name)
		if err == nil && fi.Size() == f.size {
			return true
		}
	}
	return false
}

type writeOp struct {
	buf   []byte
	off   int64
	start time.Time
	size  int64
}

func NewRootFS(ctx *Context) *RfsNode {
	rn := &RfsNode{
		ctx: ctx,
		e:   &proto.DirEntry{},
	}
	return rn
}

// resolve acts as `filepath.Join`, but we want the '/' separator always
func (rn *RfsNode) resolve(fileName string) string {
	if len(fileName) > 0 {
		p := rn.path()
		if p != string(filepath.Separator) {
			return p + string(filepath.Separator) + fileName
		}
		return string(filepath.Separator) + fileName
	}
	return rn.path()
}

func (rn *RfsNode) path() string {
	p := rn.Path(rn.Root())
	return Path(p)
}

func Path(p string) string {
	if p == "." || p == "" {
		p = string(filepath.Separator)
	} else if len(p) > 0 && p[0] != filepath.Separator && !strings.Contains(p, ignoreFile) {
		p = string(filepath.Separator) + p
	}
	return p
}

func (rn *RfsNode) cachePath(name string) string {
	return path.Join(rn.ctx.cacheDir, name)
}

func (rn *RfsNode) switchCacheFile() (wg *sync.WaitGroup) {
	if cacheFile := rn.cacheFile; cacheFile != nil && !cacheFile.immutable {
		cacheFile.immutable = true
		close(cacheFile.closeCh)
		wg = &cacheFile.wg
	}
	return
}

func (rn *RfsNode) removeCacheFile() {
	rn.mu.Lock()
	rn.switchCacheFile()
	if cacheFile := rn.cacheFile; cacheFile != nil {
		cacheFile.timer.Stop()
		cacheFile.size = 0
		cacheFile.Close()
		os.Remove(cacheFile.name)
		rn.cacheFile = nil
	}
	rn.mu.Unlock()
}

func (rn *RfsNode) removeCache() {
	if len(rn.ctx.cacheDir) > 0 {
		rn.mu.Lock()
		rn.switchCacheFile()
		cacheFile := rn.cacheFile
		rn.mu.Unlock()
		if cacheFile != nil {
			rn.ctx.cache.Lock()
			rn.ctx.cache.Remove(cacheFile.name)
			size := rn.ctx.cache.Size()
			rn.ctx.cache.Unlock()
			metrics.DiskCache.Set(float64(size))
		}
	}
}

func (rn *RfsNode) Readdir(ctx context.Context) (fs.DirStream, syscall.Errno) {
	fullPath := rn.resolve("")
	log.S().Infof("Readdir %v", fullPath)

	ctx1, cancel := context.WithTimeout(context.Background(), Timeout)
	defer cancel()

	if rn.ctx.merged {
		var dirs [][]*proto.DirEntry
		for i := range rn.ctx.clients {
			client := rn.ctx.clients[i]
			resp, err := client.Client().Readdir(ctx1, &proto.ReaddirReq{Token: rn.ctx.token, Name: fullPath})
			if err != nil {
				log.S().Error(err)
				return nil, syscall.EAGAIN
			}
			if resp.Errno != 0 {
				if resp.Errno != int64(syscall.ENOENT) {
					log.S().Warnf("Readdir %v failed: %v", fullPath, syscall.Errno(resp.Errno))
					return nil, syscall.Errno(resp.Errno)
				}
				continue
			}
			dirs = append(dirs, resp.Entries)
		}
		dirEntries := mergefs.MergeEntries(dirs...)
		if len(dirEntries) == 0 {
			return nil, syscall.ENOENT
		}
		entries := make([]fuse.DirEntry, 0, len(dirEntries))
		for _, e := range dirEntries {
			// log.S().Infof("Readdir %+v", e)
			fe := fuse.DirEntry{Mode: e.Mode, Name: e.Name, Ino: rn.inodeHash(rn.resolve(e.Name))}
			if e.IsDir {
				fe.Mode |= syscall.S_IFDIR
			} else {
				fe.Mode |= syscall.S_IFREG
			}
			entries = append(entries, fe)
		}

		return fs.NewListDirStream(entries), fs.OK
	}
	resp, err := rn.ctx.client.Client().Readdir(ctx1, &proto.ReaddirReq{Token: rn.ctx.token, Name: fullPath})
	if err != nil {
		log.S().Error(err)
		return nil, syscall.EAGAIN
	}
	if resp.Errno != 0 {
		log.S().Warnf("Readdir %v failed: %v", fullPath, syscall.Errno(resp.Errno))
		return nil, syscall.Errno(resp.Errno)
	}

	entries := make([]fuse.DirEntry, 0, len(resp.Entries))
	for _, e := range resp.Entries {
		// log.S().Infof("Readdir %+v", e)
		fe := fuse.DirEntry{Mode: e.Mode, Name: e.Name, Ino: e.Ino}
		if e.IsDir {
			fe.Mode |= syscall.S_IFDIR
		} else if e.Mode&uint32(os.ModeSymlink) > 0 {
			fe.Mode |= syscall.S_IFLNK
		} else {
			fe.Mode |= syscall.S_IFREG
		}
		entries = append(entries, fe)
	}

	return fs.NewListDirStream(entries), fs.OK
}

func (rn *RfsNode) Mknod(ctx context.Context, name string, mode uint32, dev uint32, out *fuse.EntryOut) (*fs.Inode, syscall.Errno) {
	fullPath := rn.resolve(name)
	log.S().Infof("Mknod %s, mode %d, dev %d", fullPath, mode, dev)

	if rn.ctx.merged {
		log.S().Warnf("Mknod %v err %s", fullPath, syscall.ENOSYS)
		return nil, syscall.ENOSYS
	}

	ctx1, cancel := context.WithTimeout(context.Background(), Timeout)
	defer cancel()
	resp, err := rn.ctx.client.Client().Mknod(ctx1, &proto.MknodReq{Token: rn.ctx.token, Name: fullPath, Mode: mode, Dev: dev})
	if err != nil {
		log.S().Errorf("Mknod %s, failed: %v", fullPath, err)
		return nil, syscall.EAGAIN
	}
	if resp.Errno != 0 {
		log.S().Errorf("Mknod %s, failed: %v", fullPath, syscall.Errno(resp.Errno))
		return nil, syscall.Errno(resp.Errno)
	}

	child := &RfsNode{ctx: rn.ctx, e: &proto.DirEntry{}}

	child.e.IsDir = resp.Entry.GetIsDir()
	child.e.Nlink = resp.Entry.GetNlink()
	child.e.Ino = resp.Entry.GetIno()
	// to avoid fuse driver to got wrong size
	child.e.Sz = resp.Entry.GetSz()
	child.e.ModTime = resp.Entry.GetModTime()

	log.S().Debugf("%+v", child.e)
	child.e.Mode = mode
	child.uok = rn.uok
	child.uid = rn.uid
	child.gok = rn.gok
	child.gid = rn.gid

	out.Mode = child.e.Mode
	out.Nlink = child.e.Nlink
	out.Ino = child.e.Ino
	out.Size = child.e.Sz
	out.Mtime = child.e.ModTime / 1e9
	out.Mtimensec = uint32(child.e.ModTime % 1e9)
	out.Uid = uid
	if child.uok {
		out.Uid = child.uid
	}
	out.Gid = gid
	if child.gok {
		out.Gid = child.gid
	}
	return rn.NewInode(ctx, child, fs.StableAttr{Mode: child.e.Mode, Ino: child.e.Ino, Gen: rn.inodeHash(fullPath)}), fs.OK
}

func (rn *RfsNode) Mkdir(ctx context.Context, name string, mode uint32, out *fuse.EntryOut) (*fs.Inode, syscall.Errno) {
	fullPath := rn.resolve(name)
	log.S().Infof("Mkdir %v", fullPath)

	ctx1, cancel := context.WithTimeout(context.Background(), Timeout)
	defer cancel()
	if rn.ctx.merged {
		var resp *proto.MkdirResp
		for i := range rn.ctx.clients {
			client := rn.ctx.clients[i]
			var err error
			resp, err = client.Client().Mkdir(ctx1, &proto.MkdirReq{Token: rn.ctx.token, Name: fullPath, Mode: mode})
			if err != nil {
				log.S().Error(err)
				return nil, syscall.EAGAIN
			}
			if resp.Errno != 0 {
				return nil, syscall.Errno(resp.Errno)
			}
		}

		child := &RfsNode{ctx: rn.ctx, e: &proto.DirEntry{}}

		child.e.IsDir = resp.Entry.GetIsDir()
		child.e.Mode = resp.Entry.GetMode()
		child.e.Nlink = resp.Entry.GetNlink()
		child.e.Ino = rn.inodeHash(fullPath)
		// to avoid fuse driver to got wrong size
		child.e.Sz = resp.Entry.GetSz()
		child.e.ModTime = resp.Entry.GetModTime()

		log.S().Debugf("%+v", child.e)

		child.e.Mode = mode | syscall.S_IFDIR
		child.uok = rn.uok
		child.uid = rn.uid
		child.gok = rn.gok
		child.gid = rn.gid

		out.Mode = child.e.Mode
		out.Nlink = child.e.Nlink
		out.Ino = child.e.Ino
		out.Size = child.e.Sz
		out.Mtime = child.e.ModTime / 1e9
		out.Mtimensec = uint32(child.e.ModTime % 1e9)
		out.Uid = uid
		if child.uok {
			out.Uid = child.uid
		}
		out.Gid = gid
		if child.gok {
			out.Gid = child.gid
		}
		return rn.NewInode(ctx, child, fs.StableAttr{Mode: child.e.Mode, Ino: child.e.Ino, Gen: rn.inodeHash(fullPath)}), fs.OK
	}
	resp, err := rn.ctx.client.Client().Mkdir(ctx1, &proto.MkdirReq{Token: rn.ctx.token, Name: fullPath, Mode: mode})
	if err != nil {
		log.S().Error(err)
		return nil, syscall.EAGAIN
	}
	if resp.Errno != 0 {
		return nil, syscall.Errno(resp.Errno)
	}

	child := &RfsNode{ctx: rn.ctx, e: &proto.DirEntry{}}

	child.e.IsDir = resp.Entry.GetIsDir()
	child.e.Mode = resp.Entry.GetMode()
	child.e.Nlink = resp.Entry.GetNlink()
	child.e.Ino = resp.Entry.GetIno()
	// to avoid fuse driver to got wrong size
	child.e.Sz = resp.Entry.GetSz()
	child.e.ModTime = resp.Entry.GetModTime()

	log.S().Debugf("%+v", child.e)

	child.e.Mode = mode | syscall.S_IFDIR
	child.uok = rn.uok
	child.uid = rn.uid
	child.gok = rn.gok
	child.gid = rn.gid

	out.Mode = child.e.Mode
	out.Nlink = child.e.Nlink
	out.Ino = child.e.Ino
	out.Size = child.e.Sz
	out.Mtime = child.e.ModTime / 1e9
	out.Mtimensec = uint32(child.e.ModTime % 1e9)
	out.Uid = uid
	if child.uok {
		out.Uid = child.uid
	}
	out.Gid = gid
	if child.gok {
		out.Gid = child.gid
	}
	return rn.NewInode(ctx, child, fs.StableAttr{Mode: child.e.Mode, Ino: child.e.Ino, Gen: rn.inodeHash(fullPath)}), fs.OK
}

func (rn *RfsNode) Rmdir(ctx context.Context, name string) syscall.Errno {
	fullPath := rn.resolve(name)
	log.S().Infof("Rmdir %v", fullPath)

	ctx1, cancel := context.WithTimeout(context.Background(), Timeout)
	defer cancel()
	if rn.ctx.merged {
		var resp *proto.RmdirResp
		for i := range rn.ctx.clients {
			client := rn.ctx.clients[i]
			var err error
			resp, err = client.Client().Rmdir(ctx1, &proto.RmdirReq{Token: rn.ctx.token, Name: fullPath})
			if err != nil {
				log.S().Error(err)
				return syscall.EAGAIN
			}
			if resp.Errno != 0 {
				return syscall.Errno(resp.Errno)
			}
		}
		return fs.OK
	}
	resp, err := rn.ctx.client.Client().Rmdir(ctx1, &proto.RmdirReq{Token: rn.ctx.token, Name: fullPath})
	if err != nil {
		log.S().Error(err)
		return syscall.EAGAIN
	}
	if resp.Errno != 0 {
		return syscall.Errno(resp.Errno)
	}
	return fs.OK
}

// Lookup finds a file under the current node(directory)
// TODO: cache the result of lookup to avoid repeated RPC
func (rn *RfsNode) Lookup(ctx context.Context, name string, out *fuse.EntryOut) (*fs.Inode, syscall.Errno) {
	fullPath := rn.resolve(name)
	log.S().Infof("Lookup %v", fullPath)

	ctx1, cancel := context.WithTimeout(context.Background(), Timeout)
	defer cancel()

	if rn.ctx.merged {
		rn.ctx.mergedfs.RLock()
		entry := rn.ctx.mergedfs.entries[fullPath]
		rn.ctx.mergedfs.RUnlock()
		if entry == nil {
			log.S().Warnf("Lookup %v failed: %v", fullPath, syscall.ENOENT)
			return nil, syscall.ENOENT
		}
		log.S().Debugf("Lookup %v entry: %v", fullPath, entry)

		child := &RfsNode{ctx: rn.ctx, e: &proto.DirEntry{}}

		child.e.IsDir = entry.GetIsDir()
		child.e.Mode = entry.GetMode()
		child.e.Nlink = entry.GetNlink()
		child.e.Ino = rn.inodeHash(fullPath)
		// to avoid fuse driver to got wrong size
		child.e.Sz = entry.GetSz()
		child.e.ModTime = entry.GetModTime()

		log.S().Debugf("%+v", child.e)

		if entry.IsDir {
			child.e.Mode |= syscall.S_IFDIR
		} else {
			child.e.Mode |= syscall.S_IFREG
		}
		child.uok = rn.uok
		child.uid = rn.uid
		child.gok = rn.gok
		child.gid = rn.gid

		out.Mode = child.e.Mode
		out.Nlink = child.e.Nlink
		out.Ino = child.e.Ino
		out.Size = child.e.Sz
		out.Mtime = child.e.ModTime / 1e9
		out.Mtimensec = uint32(child.e.ModTime % 1e9)
		out.Uid = uid
		if child.uok {
			out.Uid = child.uid
		}
		out.Gid = gid
		if child.gok {
			out.Gid = child.gid
		}
		return rn.NewInode(ctx, child, fs.StableAttr{Mode: child.e.Mode, Ino: child.e.Ino, Gen: rn.inodeHash(fullPath)}), fs.OK
	}
	resp, err := rn.ctx.client.Client().Lookup(ctx1, &proto.LookupReq{Token: rn.ctx.token, Name: fullPath})
	if err != nil {
		log.S().Error(err)
		return nil, syscall.EAGAIN
	}
	if resp.Errno != 0 {
		return nil, syscall.Errno(resp.Errno)
	}

	child := &RfsNode{ctx: rn.ctx, e: &proto.DirEntry{}}

	child.e.IsDir = resp.Entry.GetIsDir()
	child.e.Mode = resp.Entry.GetMode()
	child.e.Nlink = resp.Entry.GetNlink()
	child.e.Ino = resp.Entry.GetIno()
	// to avoid fuse driver to got wrong size
	child.e.Sz = resp.Entry.GetSz()
	child.e.ModTime = resp.Entry.GetModTime()

	log.S().Debugf("%+v", child.e)

	if resp.Entry.IsDir {
		child.e.Mode |= syscall.S_IFDIR
	} else if resp.Entry.GetMode()&uint32(os.ModeSymlink) > 0 {
		child.e.Mode |= syscall.S_IFLNK
	} else {
		child.e.Mode |= syscall.S_IFREG
	}
	child.uok = rn.uok
	child.uid = rn.uid
	child.gok = rn.gok
	child.gid = rn.gid

	out.Mode = child.e.Mode
	out.Nlink = child.e.Nlink
	out.Ino = child.e.Ino
	out.Size = child.e.Sz
	out.Mtime = child.e.ModTime / 1e9
	out.Mtimensec = uint32(child.e.ModTime % 1e9)
	out.Uid = uid
	if child.uok {
		out.Uid = child.uid
	}
	out.Gid = gid
	if child.gok {
		out.Gid = child.gid
	}
	return rn.NewInode(ctx, child, fs.StableAttr{Mode: child.e.Mode, Ino: child.e.Ino, Gen: rn.inodeHash(fullPath)}), fs.OK
}

func (rn *RfsNode) Create(ctx context.Context, name string, flags uint32, mode uint32, out *fuse.EntryOut) (node *fs.Inode, fh fs.FileHandle, fuseFlags uint32, errno syscall.Errno) {
	fullPath := rn.resolve(name)

	ctx1, cancel := context.WithTimeout(context.Background(), Timeout)
	defer cancel()
	fileHandle := &proto.FileHandle{
		Token: rn.ctx.token,
		Name:  fullPath,
		Flags: flags,
	}
	req := &proto.CreateReq{
		FileHandle: fileHandle,
		Mode:       mode | syscall.S_IFREG,
	}

	var resp *proto.CreateResp
	var err error
	if rn.ctx.merged {
		for i := 0; i < len(rn.ctx.clients); i++ {
			client := rn.ctx.Client()
			resp, err = client.Client().Create(ctx1, req)
			if err == nil && resp.Errno == 0 {
				break
			}
			rn.ctx.Next()
		}
	} else {
		resp, err = rn.ctx.client.Client().Create(ctx1, req)
	}

	if err != nil {
		log.S().Errorf("Create %s failed: %v", fullPath, err)
		return nil, nil, 0, fs.ToErrno(err)
	}
	if resp.Errno != 0 {
		log.S().Errorf("Create %s failed: %v", fullPath, syscall.Errno(resp.Errno))
		return nil, nil, 0, syscall.Errno(resp.Errno)
	}

	if rn.ctx.merged {
		entry := resp.Entry
		if entry != nil {
			rn.ctx.mergedfs.Lock()
			rn.ctx.mergedfs.entries[fullPath] = entry
			rn.ctx.mergedfs.Unlock()
		}
	}

	child := &RfsNode{ctx: rn.ctx, e: &proto.DirEntry{}}

	child.e.IsDir = resp.Entry.GetIsDir()
	child.e.Mode = resp.Entry.GetMode()
	child.e.Nlink = resp.Entry.GetNlink()
	if rn.ctx.merged {
		child.e.Ino = rn.inodeHash(fullPath)
	} else {
		child.e.Ino = resp.Entry.GetIno()
	}
	child.e.Sz = resp.Entry.GetSz()
	child.e.ModTime = resp.Entry.GetModTime()
	if resp.Entry.IsDir {
		child.e.Mode = mode | syscall.S_IFDIR
	} else {
		child.e.Mode = mode | syscall.S_IFREG
	}
	child.uok = rn.uok
	child.uid = rn.uid
	child.gok = rn.gok
	child.gid = rn.gid

	out.Mode = child.e.Mode
	out.Nlink = child.e.Nlink
	out.Ino = child.e.Ino
	out.Size = child.e.Sz
	out.Mtime = child.e.ModTime / 1e9
	out.Mtimensec = uint32(child.e.ModTime % 1e9)
	out.Uid = uid
	if child.uok {
		out.Uid = child.uid
	}
	out.Gid = gid
	if child.gok {
		out.Gid = child.gid
	}

	if err := child.openStream(); err != nil {
		log.S().Errorf("Open %s stream failed: %v", fullPath, err)
		return nil, nil, 0, fs.ToErrno(err)
	}
	child.fhs = make(map[fileHandleKey]int32)
	child.fhs[newFileHandleKey(fileHandle)] = 1
	child.fullPath = fullPath
	if len(child.ctx.cacheDir) > 0 {
		cachePath := child.cachePath(fullPath)
		log.S().Debugf("Create cache %s for %s, dir %s", cachePath, fullPath, filepath.Dir(cachePath))
		if err := os.MkdirAll(filepath.Dir(cachePath), os.ModePerm); err == nil {
			f, err := os.OpenFile(cachePath, int(flags), os.FileMode(mode))
			if err != nil {
				log.S().Warnf("Create cache %s failed, err:%v", fullPath, err)
			} else {
				log.S().Infof("Create cache %s successfully", cachePath)
				cacheFile := &cacheFile{
					File:       f,
					appendMode: int(flags)&syscall.O_APPEND != 0,
					name:       cachePath,
					writes:     make(chan writeOp, 1024),
					closeCh:    make(chan struct{}),
					timer:      time.AfterFunc(time.Minute*10, child.removeCache),
				}
				child.cacheFile = cacheFile
				cacheFile.wg.Add(1)
				go child.writeLoop()
			}
		}
	}
	fuseFlags = fuse.FOPEN_DIRECT_IO
	if child.ctx.keepCache {
		fuseFlags = fuse.FOPEN_KEEP_CACHE
	}
	// 33188 => 644
	node = rn.NewInode(ctx, child, fs.StableAttr{Mode: mode, Ino: child.e.Ino, Gen: rn.inodeHash(fullPath)})
	log.S().Infof("Create %v, flags:%v, mode: %v, ino: %d, file handle %v %p %p", fullPath, flags, mode, child.e.Ino, fileHandle, child, node)
	return node, fileHandle, fuseFlags, fs.OK
}

func (rn *RfsNode) Rename(ctx context.Context, name string, newParent fs.InodeEmbedder, newName string, flags uint32) syscall.Errno {
	newParentNode := (*RfsNode)(unsafe.Pointer(newParent.EmbeddedInode()))
	fullPath := rn.resolve(name)
	newFullPath := newParentNode.resolve(newName)
	log.S().Infof("Rename %v, %v", fullPath, newFullPath)

	if rn.ctx.merged {
		log.S().Warnf("Rename %v, %v, err %v", fullPath, newFullPath, syscall.ENOSYS)
		return syscall.ENOSYS
	}

	ctx1, cancel := context.WithTimeout(context.Background(), Timeout)
	defer cancel()
	resp, err := rn.ctx.client.Client().Rename(ctx1, &proto.RenameReq{Token: rn.ctx.token, NewName: newFullPath, OldName: fullPath})
	if err != nil {
		log.S().Errorf("Rename %s failed: %v", fullPath, err)
		return fs.ToErrno(err)
	}
	if resp.Errno != 0 {
		log.S().Errorf("Rename %s failed: %v", fullPath, syscall.Errno(resp.Errno))
		return syscall.Errno(resp.Errno)
	}

	child := (*RfsNode)(unsafe.Pointer(rn.GetChild(name).EmbeddedInode()))
	child.removeCache()
	child.fullPath = ""

	return fs.OK
}

func (rn *RfsNode) Link(ctx context.Context, target fs.InodeEmbedder, name string, out *fuse.EntryOut) (node *fs.Inode, errno syscall.Errno) {
	targetNode := (*RfsNode)(unsafe.Pointer(target.EmbeddedInode()))
	fullPath := targetNode.path()
	newFullPath := rn.resolve(name)
	log.S().Infof("Link %v, %v", fullPath, newFullPath)

	if rn.ctx.merged {
		log.S().Warnf("Link %v, %v, err %v", fullPath, newFullPath, syscall.ENOSYS)
		return nil, syscall.ENOSYS
	}

	ctx1, cancel := context.WithTimeout(context.Background(), Timeout)
	defer cancel()
	resp, err := rn.ctx.client.Client().Link(ctx1, &proto.LinkReq{Token: rn.ctx.token, NewName: newFullPath, OldName: fullPath})
	if err != nil {
		log.S().Errorf("Link %s failed: %v", fullPath, err)
		return nil, fs.ToErrno(err)
	}
	if resp.Errno != 0 {
		log.S().Errorf("Link %s failed: %v", fullPath, syscall.Errno(resp.Errno))
		return nil, syscall.Errno(resp.Errno)
	}
	targetNode.e.Nlink++
	targetNode.removeCache()
	targetNode.fullPath = ""
	out.Mode = targetNode.e.Mode
	out.Nlink = targetNode.e.Nlink
	out.Ino = targetNode.e.Ino
	out.Size = targetNode.e.Sz
	out.Mtime = targetNode.e.ModTime / 1e9
	out.Mtimensec = uint32(targetNode.e.ModTime % 1e9)
	out.Uid = uid
	if targetNode.uok {
		out.Uid = targetNode.uid
	}
	out.Gid = gid
	if targetNode.gok {
		out.Gid = targetNode.gid
	}
	child := &RfsNode{ctx: rn.ctx, e: targetNode.e, uid: targetNode.uid, uok: targetNode.uok, gid: targetNode.gid, gok: targetNode.gok}
	node = rn.NewInode(ctx, child, fs.StableAttr{Mode: targetNode.e.Mode, Ino: targetNode.e.Ino, Gen: rn.inodeHash(newFullPath)})
	log.S().Infof("Link target %p, targetNode %p, child %p, node %p, rn %p", target, targetNode, child, node, rn)
	return node, fs.OK
}

func (rn *RfsNode) Symlink(ctx context.Context, target, name string, out *fuse.EntryOut) (node *fs.Inode, errno syscall.Errno) {
	fullPath := target
	newFullPath := rn.resolve(name)
	log.S().Infof("Symlink %v, %v", fullPath, newFullPath)

	if rn.ctx.merged {
		log.S().Warnf("Symlink %v, %v, err %v", fullPath, newFullPath, syscall.ENOSYS)
		return nil, syscall.ENOSYS
	}

	ctx1, cancel := context.WithTimeout(context.Background(), Timeout)
	defer cancel()
	resp, err := rn.ctx.client.Client().Symlink(ctx1, &proto.SymlinkReq{Token: rn.ctx.token, NewName: newFullPath, OldName: fullPath})
	if err != nil {
		log.S().Errorf("Symlink %s failed: %v", fullPath, err)
		return nil, fs.ToErrno(err)
	}
	if resp.Errno != 0 {
		log.S().Errorf("Symlink %s failed: %v", fullPath, syscall.Errno(resp.Errno))
		return nil, syscall.Errno(resp.Errno)
	}

	child := &RfsNode{ctx: rn.ctx, e: &proto.DirEntry{}}

	child.e.IsDir = resp.Entry.GetIsDir()
	child.e.Mode = resp.Entry.GetMode()
	child.e.Nlink = resp.Entry.GetNlink()
	child.e.Ino = resp.Entry.GetIno()
	// to avoid fuse driver to got wrong size
	child.e.Sz = resp.Entry.GetSz()
	child.e.ModTime = resp.Entry.GetModTime()

	log.S().Debugf("%+v", child.e)
	child.e.Mode |= syscall.S_IFLNK
	child.uok = rn.uok
	child.uid = rn.uid
	child.gok = rn.gok
	child.gid = rn.gid

	out.Mode = child.e.Mode
	out.Nlink = child.e.Nlink
	out.Ino = child.e.Ino
	out.Size = child.e.Sz
	out.Mtime = child.e.ModTime / 1e9
	out.Mtimensec = uint32(child.e.ModTime % 1e9)
	out.Uid = uid
	if child.uok {
		out.Uid = child.uid
	}
	out.Gid = gid
	if child.gok {
		out.Gid = child.gid
	}
	return rn.NewInode(ctx, child, fs.StableAttr{Mode: child.e.Mode, Ino: child.e.Ino, Gen: rn.inodeHash(newFullPath)}), fs.OK
}

func (rn *RfsNode) Unlink(ctx context.Context, name string) syscall.Errno {
	fullPath := rn.resolve(name)
	log.S().Infof("Unlink %v", fullPath)

	ctx1, cancel := context.WithTimeout(context.Background(), Timeout)
	defer cancel()

	var resp *proto.UnlinkResp
	var err error
	if rn.ctx.merged {
		var resps = make([]*proto.UnlinkResp, len(rn.ctx.clients))
		var errs = make([]error, len(rn.ctx.clients))
		for i := 0; i < len(rn.ctx.clients); i++ {
			client := rn.ctx.clients[i]
			resps[i], errs[i] = client.Client().Unlink(ctx1, &proto.UnlinkReq{Token: rn.ctx.token, Name: fullPath})
			realm := rn.ctx.realms[i]
			if errs[i] != nil {
				log.S().Warnf("Unlink %s %s %s failed: %v", fullPath, realm.Name, realm.Address, errs[i])
			}
		}
		for i := 0; i < len(rn.ctx.clients); i++ {
			if errs[i] == nil {
				resp = resps[i]
				err = nil
				break
			}
			err = errs[i]
		}
	} else {
		resp, err = rn.ctx.Client().Client().Unlink(ctx1, &proto.UnlinkReq{Token: rn.ctx.token, Name: fullPath})
	}
	if err != nil {
		log.S().Errorf("Unlink %s failed: %v", fullPath, err)
		return fs.ToErrno(err)
	}
	if resp.Errno != 0 {
		log.S().Errorf("Unlink %s failed: %v", fullPath, syscall.Errno(resp.Errno))
		return syscall.Errno(resp.Errno)
	}
	if rn.ctx.merged {
		rn.ctx.mergedfs.Lock()
		delete(rn.ctx.mergedfs.entries, fullPath)
		rn.ctx.mergedfs.Unlock()
		log.S().Infof("delete %v entry", fullPath)
	}

	child := (*RfsNode)(unsafe.Pointer(rn.GetChild(name).EmbeddedInode()))
	child.e.Nlink--
	child.removeCache()
	child.fullPath = ""

	return syscall.Errno(resp.Errno)
}

func (rn *RfsNode) Statfs(ctx context.Context, out *fuse.StatfsOut) syscall.Errno {
	fullPath := rn.path()
	log.S().Debugf("Statfs %v", fullPath)

	if rn.ctx.merged {
		rn.ctx.mergedfs.RLock()
		entry := rn.ctx.mergedfs.entries[fullPath]
		rn.ctx.mergedfs.RUnlock()
		if entry == nil {
			log.S().Warnf("Statfs %v failed: %v", fullPath, syscall.ENOENT)
			return syscall.ENOENT
		}
	}
	ctx1, cancel := context.WithTimeout(context.Background(), Timeout)
	defer cancel()
	resp, err := rn.ctx.client.Client().Statfs(ctx1, &proto.StatfsReq{Token: rn.ctx.token, Name: fullPath})
	if err != nil {
		log.S().Errorf("Statfs %s failed: %v", fullPath, err)
		return fs.ToErrno(err)
	}
	convertToStatfsOut(resp, out)
	log.S().Debugf("Statfs %v", *out)
	return fs.OK
}

// Hash file path into inode number, so we can ensure the same file always gets the same inode number
func (n *RfsNode) inodeHash(path string) uint64 {
	h := fnv.New64a()
	_, _ = h.Write([]byte(path))
	return h.Sum64()
}