package client

import (
	"context"
	"io"
	"os"
	"strings"
	"sync"
	"syscall"
	"time"

	"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"
	"github.com/tidbcloud/ffs/util/logutil"
)

const (
	MaxKernelWrite = 128 * 1024
	Timeout        = time.Minute * 2
)

var bufferPool = &sync.Pool{
	New: func() interface{} {
		return make([]byte, MaxKernelWrite)
	},
}

func getBuffer() []byte {
	return bufferPool.Get().([]byte)
}

func putBuffer(buf []byte) {
	if cap(buf) == MaxKernelWrite {
		buf = buf[:MaxKernelWrite]
		bufferPool.Put(buf)
	}
}

var (
	_ fs.NodeUnlinker   = &RfsNode{}
	_ fs.NodeCreater    = &RfsNode{}
	_ fs.NodeOpener     = &RfsNode{}
	_ fs.NodeReader     = &RfsNode{}
	_ fs.NodeWriter     = &RfsNode{}
	_ fs.NodeFlusher    = &RfsNode{}
	_ fs.NodeFsyncer    = &RfsNode{}
	_ fs.NodeSetattrer  = &RfsNode{}
	_ fs.NodeGetattrer  = &RfsNode{}
	_ fs.NodeReleaser   = &RfsNode{}
	_ fs.NodeRenamer    = &RfsNode{}
	_ fs.NodeLookuper   = &RfsNode{}
	_ fs.NodeMkdirer    = &RfsNode{}
	_ fs.NodeRmdirer    = &RfsNode{}
	_ fs.NodeStatfser   = &RfsNode{}
	_ fs.NodeLinker     = &RfsNode{}
	_ fs.NodeMknoder    = &RfsNode{}
	_ fs.NodeSymlinker  = &RfsNode{}
	_ fs.NodeReadlinker = &RfsNode{}
	_ fs.NodeAccesser   = &RfsNode{}
	_ fs.NodeAllocater  = &RfsNode{}
)

func (rn *RfsNode) Open(ctx context.Context, flags uint32) (fh fs.FileHandle, fuseFlags uint32, errno syscall.Errno) {
	startTime := time.Now()
	fullPath := rn.path()
	if rn.e.Mode == 0 {
		return nil, 0, syscall.EACCES
	}
	ctx1, cancel := context.WithTimeout(context.Background(), Timeout)
	defer cancel()
	fileHandle := &proto.FileHandle{
		Token: rn.ctx.token,
		Name:  fullPath,
		Flags: flags,
	}

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

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

	rn.mu.Lock()
	if rn.fhs == nil {
		rn.fhs = make(map[fileHandleKey]int32)
	}
	key := newFileHandleKey(fileHandle)

	if n, ok := rn.fhs[key]; ok {
		rn.fhs[key] = n + 1
	} else {
		rn.fhs[key] = 1
	}

	numFds := int32(0)
	for _, n := range rn.fhs {
		numFds += n
	}
	if numFds == 1 {
		log.S().Infof("Open %s Stream file handle %v", fullPath, fileHandle)
		if err := rn.openStream(); err != nil {
			rn.mu.Unlock()
			log.S().Errorf("Open %s stream failed: %v", fullPath, err)
			return nil, 0, fs.ToErrno(err)
		}
	}
	rn.fullPath = fullPath
	cacheFile := rn.cacheFile
	var wg *sync.WaitGroup
	if len(rn.ctx.cacheDir) > 0 {
		wg = rn.switchCacheFile()
	}
	rn.mu.Unlock()
	if wg != nil {
		wg.Wait()
	}

	var cached bool
	if len(rn.ctx.cacheDir) > 0 {
		rn.mu.Lock()
		cacheFile = rn.cacheFile
		if cacheFile != nil {
			if cacheFile.cached() {
				f, err := os.OpenFile(cacheFile.name, int(flags), os.ModePerm)
				if err != nil {
					log.S().Warnf("Open cache %s failed, err:%v", cacheFile.name, err)
				} else {
					cacheFile.File = f
					cached = true
				}
			}
		}
		rn.mu.Unlock()
	}
	if cacheFile != nil {
		rn.ctx.cache.Lock()
		if !cached {
			rn.ctx.cache.Remove(cacheFile.name)
		}
		size := rn.ctx.cache.Size()
		rn.ctx.cache.Unlock()
		metrics.DiskCache.Set(float64(size))
	}

	log.S().Infof("Open %s, flags %d, file handle %v %p", fullPath, flags, fileHandle, rn)
	metrics.OpenWaitTimeDurationHistogram.Observe(time.Since(startTime).Seconds())
	fuseFlags = fuse.FOPEN_DIRECT_IO
	if rn.ctx.keepCache {
		fuseFlags = fuse.FOPEN_KEEP_CACHE
	}
	return fileHandle, fuseFlags, fs.OK
}

func (rn *RfsNode) openStream() (err error) {
	if rn.ctx.merged {
		stream, err := newBatchCommandsClient(func(ctx context.Context) (proto.RealmService_BatchCommandsClient, error) {
			return rn.ctx.Client().Client().BatchCommands(ctx)
		})
		if err != nil {
			return err
		}
		rn.stream = stream
		return nil
	}
	stream, err := newBatchCommandsClient(func(ctx context.Context) (proto.RealmService_BatchCommandsClient, error) {
		return rn.ctx.client.Client().BatchCommands(ctx)
	})
	if err != nil {
		return err
	}
	rn.stream = stream
	return nil
}

func getFileHandle(fh fs.FileHandle) *proto.FileHandle {
	if fileHandle, ok := fh.(*proto.FileHandle); ok {
		return fileHandle
	}
	return nil
}

func (rn *RfsNode) Read(ctx context.Context, fh fs.FileHandle, dest []byte, off int64) (fuse.ReadResult, syscall.Errno) {
	startTime := time.Now()
	fullPath := rn.fullPath
	if len(fullPath) == 0 {
		fullPath = rn.path()
		rn.fullPath = fullPath
	}
	log.S().Debugf("Read %s, off %d, len %d, file handle %v", fullPath, off, len(dest), fh)
	cacheFile := rn.cacheFile
	if len(rn.ctx.cacheDir) > 0 && cacheFile != nil && cacheFile.immutable {
		if cacheFile.size >= off+int64(len(dest)) {
			n, err := cacheFile.ReadAt(dest, off)
			if err != nil && err != io.EOF {
				log.S().Warnf("Read cache %s, err: %v", fullPath, err)
			} else {
				metrics.ReadCacheSizeByteHistogram.Observe(float64(len(dest)))
				metrics.ReadCacheWaitTimeDurationHistogram.Observe(time.Since(startTime).Seconds())
				rn.ctx.cache.Lock()
				if _, r, ok := rn.ctx.cache.Get(cacheFile.name); ok {
					r.Done()
				}
				rn.ctx.cache.Unlock()
				return fuse.ReadResultData(dest[:n]), fs.OK
			}
		}
	}
	if rn.ctx.merged {
		var frames []*proto.Frame
		req := &proto.ReadReq{FileHandle: getFileHandle(fh), Off: off, Len: int64(len(dest))}
		for i := range rn.ctx.clients {
			client := rn.ctx.clients[i]
			resp, err := client.Client().Read(ctx, req)
			if err != nil {
				log.S().Error(err)
				return nil, syscall.EAGAIN
			}
			if resp.Errno != 0 {
				if resp.Errno != int64(syscall.ENOENT) {
					log.S().Warnf("Read %v failed: %v", fullPath, syscall.Errno(resp.Errno))
					return nil, syscall.Errno(resp.Errno)
				}
				continue
			}
			realm := rn.ctx.realms[i]
			log.S().Infof("Read %v from %s %s frames: %v", fullPath, realm.Name, realm.Address, len(resp.Frames))
			frames = mergefs.MergeFrames(frames, resp.Frames)
		}
		if len(frames) == 0 {
			log.S().Warnf("Read %v failed: %v", fullPath, syscall.ENOENT)
			return nil, syscall.ENOENT
		}
		log.S().Infof("Read %v frames: %v", fullPath, len(frames))
		n := mergefs.ReadAt(frames, dest, off)
		metrics.ReadSizeByteHistogram.Observe(float64(len(dest)))
		metrics.ReadWaitTimeDurationHistogram.Observe(time.Since(startTime).Seconds())
		return fuse.ReadResultData(dest[:n]), fs.OK
	}
	req := &proto.ReadReq{FileHandle: getFileHandle(fh), Off: off, Len: int64(len(dest))}
	cb := newCallback(&proto.BatchCommandsRequest_Request{Cmd: &proto.BatchCommandsRequest_Request_Read{Read: req}})
	if err := rn.stream.Send(cb); err != nil {
		log.S().Errorf("Read %s send failed: %v", fullPath, err.Error())
		return nil, syscall.EAGAIN
	}

	res := cb.Wait()

	switch res := res.GetCmd().(type) {
	case *proto.BatchCommandsResponse_Response_Read:
		resp := res.Read
		if resp.Errno != 0 {
			log.S().Errorf("Read %s failed: %v", fullPath, syscall.Errno(resp.Errno))
			return nil, syscall.Errno(resp.Errno)
		}
		if rn.ctx.merged {
			n := mergefs.ReadAt(resp.Frames, dest, off)
			metrics.ReadSizeByteHistogram.Observe(float64(len(dest)))
			metrics.ReadWaitTimeDurationHistogram.Observe(time.Since(startTime).Seconds())
			return fuse.ReadResultData(dest[:n]), fs.OK
		}
		buf := resp.GetBuf()
		copy(dest, buf)

		metrics.ReadSizeByteHistogram.Observe(float64(len(buf)))
		metrics.ReadWaitTimeDurationHistogram.Observe(time.Since(startTime).Seconds())
		return fuse.ReadResultData(dest[:len(buf)]), fs.OK
	}
	return nil, syscall.EAGAIN
}

// only support append file!!!!
func (rn *RfsNode) Write(ctx context.Context, fh fs.FileHandle, buf []byte, off int64) (uint32, syscall.Errno) {
	startTime := time.Now()
	fullPath := rn.fullPath
	if len(fullPath) == 0 {
		fullPath = rn.path()
		rn.fullPath = fullPath
	}
	sz := int64(len(buf))
	log.S().Debugf("Write %s, offset:%d, sz:%d, file handle %v", fullPath, off, sz, fh)
	clone := getBuffer()[:len(buf)]
	copy(clone, buf)
	req := &proto.WriteReq{FileHandle: getFileHandle(fh), Off: off, Buf: clone}
	cb := newCallback(&proto.BatchCommandsRequest_Request{Cmd: &proto.BatchCommandsRequest_Request_Write{Write: req}})

	if rn.ctx.asyncWrite {
		cb.doneFn = func() {
			res := cb.Wait()
			switch res := res.GetCmd().(type) {
			case *proto.BatchCommandsResponse_Response_Write:
				resp := res.Write
				if resp.Errno != 0 {
					log.S().Errorf("Write %s failed: %v", fullPath, syscall.Errno(resp.Errno))
					return
				}
				putBuffer(clone)
				metrics.WriteWaitTimeDurationHistogram.Observe(time.Since(startTime).Seconds())
				return
			}
		}
		if err := rn.stream.Send(cb); err != nil {
			log.S().Errorf("Write %s send failed: %v", fullPath, err.Error())
			return 0, syscall.EAGAIN
		}
	} else {
		if err := rn.stream.Send(cb); err != nil {
			log.S().Errorf("Write %s send failed: %v", fullPath, err.Error())
			return 0, syscall.EAGAIN
		}
		res := cb.Wait()
		switch res := res.GetCmd().(type) {
		case *proto.BatchCommandsResponse_Response_Write:
			resp := res.Write
			if resp.Errno != 0 {
				log.S().Errorf("Write %s failed: %v", fullPath, syscall.Errno(resp.Errno))
				return 0, syscall.Errno(resp.Errno)
			}
			putBuffer(clone)
			metrics.WriteWaitTimeDurationHistogram.Observe(time.Since(startTime).Seconds())
		}
	}

	cacheFile := rn.cacheFile
	if len(rn.ctx.cacheDir) > 0 && cacheFile != nil && !cacheFile.immutable {
		if cacheFile.size == off {
			cacheFile.size += int64(len(buf))
			cp := getBuffer()[:len(buf)]
			copy(cp, buf)
			select {
			case cacheFile.writes <- writeOp{buf: cp, off: off, start: startTime}:
			case <-cacheFile.closeCh:
			}
		}
	}
	if rn.e.GetSz() < uint64(off+sz) {
		rn.e.Sz = uint64(off + sz)
	}

	if rn.ctx.merged {
		rn.ctx.mergedfs.RLock()
		entry := rn.ctx.mergedfs.entries[fullPath]
		rn.ctx.mergedfs.RUnlock()
		if entry != nil {
			if entry.Sz < rn.e.Sz {
				entry.Sz = rn.e.Sz
			}
		}
		log.S().Debugf("change %v entry: %v", fullPath, entry)
	}
	metrics.WriteSizeByteHistogram.Observe(float64(len(clone)))
	return uint32(sz), fs.OK
}

func (rn *RfsNode) write(op writeOp) {
	cacheFile := rn.cacheFile
	if cacheFile != nil {
		if len(op.buf) > 0 {
			rn.ctx.cache.Lock()
			r, ok := rn.ctx.cache.Set(cacheFile.name, rn, int(op.off)+len(op.buf))
			r.Done()
			size := rn.ctx.cache.Size()
			rn.ctx.cache.Unlock()
			if !ok {
				log.S().Warnf("Write cache %s failed: too big %d", cacheFile.name, int(op.off)+len(op.buf))
				rn.removeCacheFile()
			}
			metrics.DiskCache.Set(float64(size))
			var n int
			var err error
			if cacheFile.appendMode {
				n, err = cacheFile.Write(op.buf)
			} else {
				n, err = cacheFile.WriteAt(op.buf, op.off)
			}
			if n == len(op.buf) {
				metrics.WriteCacheSizeByteHistogram.Observe(float64(len(op.buf)))
				metrics.WriteCacheWaitTimeDurationHistogram.Observe(time.Since(op.start).Seconds())
			} else if err != nil {
				log.S().Warnf("Write cache %s failed: %v", cacheFile.name, err.Error())
			}
		} else {
			rn.ctx.cache.Lock()
			r, ok := rn.ctx.cache.Set(cacheFile.name, rn, int(op.size))
			r.Done()
			size := rn.ctx.cache.Size()
			rn.ctx.cache.Unlock()
			if !ok {
				log.S().Warnf("Truncate cache %s failed: too big %d", cacheFile.name, op.size)
				rn.removeCacheFile()
			}
			metrics.DiskCache.Set(float64(size))
			err := cacheFile.Truncate(op.size)
			if err != nil {
				log.S().Warnf("Truncate cache %s failed: %v", cacheFile.name, err.Error())
			}
		}
	}
	buf := op.buf
	putBuffer(buf)
}

func (rn *RfsNode) writeLoop() {
	cacheFile := rn.cacheFile
	defer cacheFile.wg.Done()
	for {
		select {
		case <-cacheFile.closeCh:
			pending := len(cacheFile.writes)
			for i := 0; i < pending; i++ {
				op := <-cacheFile.writes
				rn.write(op)
			}
			return
		case op := <-cacheFile.writes:
			rn.write(op)
		}
	}
}

func (rn *RfsNode) Flush(ctx context.Context, fh fs.FileHandle) syscall.Errno {
	startTime := time.Now()
	fullPath := rn.fullPath
	if len(fullPath) == 0 {
		fullPath = rn.path()
		rn.fullPath = fullPath
	}
	log.S().Infof("Flush %v, file handle %v", fullPath, fh)

	req := &proto.FlushReq{FileHandle: getFileHandle(fh)}
	cb := newCallback(&proto.BatchCommandsRequest_Request{Cmd: &proto.BatchCommandsRequest_Request_Flush{Flush: req}})
	if err := rn.stream.Send(cb); err != nil {
		log.S().Errorf("Flush %s send failed: %v", fullPath, err.Error())
		return syscall.EAGAIN
	}

	res := cb.Wait()

	switch res := res.GetCmd().(type) {
	case *proto.BatchCommandsResponse_Response_Flush:
		resp := res.Flush
		if resp.Errno != 0 {
			if resp.Errno != int64(syscall.ENOENT) {
				log.S().Errorf("Flush %s failed: %v", fullPath, syscall.Errno(resp.Errno))
			}
			return syscall.Errno(resp.Errno)
		}
		if rn.ctx.merged {
			if rn.e.Sz < resp.Entry.Sz {
				rn.e.Sz = resp.Entry.Sz
			}
			if rn.e.ModTime < resp.Entry.ModTime {
				rn.e.ModTime = resp.Entry.ModTime
			}
			rn.ctx.mergedfs.RLock()
			entry := rn.ctx.mergedfs.entries[fullPath]
			rn.ctx.mergedfs.RUnlock()
			if entry != nil {
				if entry.Sz < rn.e.Sz {
					entry.Sz = rn.e.Sz
				}
				if entry.ModTime < rn.e.ModTime {
					entry.ModTime = rn.e.ModTime
				}
			}
			log.S().Debugf("change %v entry: %v", fullPath, entry)
		} else {
			rn.e.Sz = resp.GetEntry().GetSz()
			rn.e.ModTime = resp.GetEntry().GetModTime()
		}
		log.S().Infof("Flush %s, %v", fullPath, rn.e.Sz)
		metrics.FlushWaitTimeDurationHistogram.Observe(time.Since(startTime).Seconds())

		return fs.OK
	}
	return syscall.EAGAIN
}

func (rn *RfsNode) Fsync(ctx context.Context, fh fs.FileHandle, flags uint32) syscall.Errno {
	startTime := time.Now()
	fullPath := rn.fullPath
	if len(fullPath) == 0 {
		fullPath = rn.path()
		rn.fullPath = fullPath
	}

	log.S().Infof("Fsync %v, file handle %v", fullPath, fh)

	req := &proto.FsyncReq{FileHandle: getFileHandle(fh)}
	if req.FileHandle == nil {
		req.FileHandle = &proto.FileHandle{
			Token: rn.ctx.token,
			Name:  fullPath,
			Flags: flags,
		}
		resp, err := rn.ctx.Client().Client().Fsync(ctx, req)
		if err != nil {
			log.S().Errorf("Fsync %s failed: %v", fullPath, err)
			return fs.ToErrno(err)
		}
		if resp.Errno != 0 {
			log.S().Errorf("Fsync %s failed: %v", fullPath, syscall.Errno(resp.Errno))
			return syscall.Errno(resp.Errno)
		}
		rn.e.Sz = resp.GetEntry().GetSz()
		rn.e.ModTime = resp.GetEntry().GetModTime()
		log.S().Infof("Fsync %s", fullPath)
		metrics.FsyncWaitTimeDurationHistogram.Observe(time.Since(startTime).Seconds())
		return fs.OK
	}
	cb := newCallback(&proto.BatchCommandsRequest_Request{Cmd: &proto.BatchCommandsRequest_Request_Fsync{Fsync: req}})
	if err := rn.stream.Send(cb); err != nil {
		log.S().Errorf("Fsync %s send failed: %v", fullPath, err.Error())
		return syscall.EAGAIN
	}

	res := cb.Wait()

	switch res := res.GetCmd().(type) {
	case *proto.BatchCommandsResponse_Response_Fsync:
		resp := res.Fsync
		if resp.Errno != 0 {
			log.S().Errorf("Fsync %s failed: %v", fullPath, syscall.Errno(resp.Errno))
			return syscall.Errno(resp.Errno)
		}
		if rn.ctx.merged {
			if rn.e.Sz < resp.Entry.Sz {
				rn.e.Sz = resp.Entry.Sz
			}
			if rn.e.ModTime < resp.Entry.ModTime {
				rn.e.ModTime = resp.Entry.ModTime
			}
			rn.ctx.mergedfs.RLock()
			entry := rn.ctx.mergedfs.entries[fullPath]
			rn.ctx.mergedfs.RUnlock()
			if entry != nil {
				if entry.Sz < rn.e.Sz {
					entry.Sz = rn.e.Sz
				}
				if entry.ModTime < rn.e.ModTime {
					entry.ModTime = rn.e.ModTime
				}
			}
			log.S().Debugf("change %v entry: %v", fullPath, entry)
		} else {
			rn.e.Sz = resp.GetEntry().GetSz()
			rn.e.ModTime = resp.GetEntry().GetModTime()
		}

		log.S().Infof("Fsync %s", fullPath)
		metrics.FsyncWaitTimeDurationHistogram.Observe(time.Since(startTime).Seconds())

		return fs.OK
	}

	return syscall.EAGAIN
}

func (rn *RfsNode) Release(ctx context.Context, fh fs.FileHandle) syscall.Errno {
	rn.Flush(ctx, fh)
	fullPath := rn.path()
	rn.fullPath = ""
	log.S().Infof("Release %v, file handle %v", fullPath, fh)
	ctx1, cancel := context.WithTimeout(context.Background(), Timeout)
	defer cancel()
	fileHandle := getFileHandle(fh)
	var resp *proto.CloseResp
	var err error
	if rn.ctx.merged {
		var resps = make([]*proto.CloseResp, 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().Close(ctx1, &proto.CloseReq{FileHandle: fileHandle})
			realm := rn.ctx.realms[i]
			if errs[i] != nil {
				log.S().Warnf("Release %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().Close(ctx1, &proto.CloseReq{FileHandle: fileHandle})

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

	rn.mu.Lock()
	var wg *sync.WaitGroup
	var size int
	if len(rn.ctx.cacheDir) > 0 {
		wg = rn.switchCacheFile()
		size = rn.ctx.cache.Size()
	}

	key := newFileHandleKey(fileHandle)
	if n, ok := rn.fhs[key]; ok {
		if n-1 > 0 {
			rn.fhs[key] = n - 1
		} else {
			delete(rn.fhs, key)
		}
	}

	numFds := int32(0)
	for _, n := range rn.fhs {
		numFds += n
	}
	if numFds == 0 {
		log.S().Infof("Close %v stream, file handle %v", fullPath, fh)
		stream := rn.stream
		if stream != nil {
			stream.close()
		}
	}
	if len(rn.fhs) == 0 {
		rn.fhs = nil
	}
	rn.mu.Unlock()
	if wg != nil {
		wg.Wait()
		metrics.DiskCache.Set(float64(size))
	}
	return fs.OK
}

func (rn *RfsNode) Setattr(ctx context.Context, fh fs.FileHandle, in *fuse.SetAttrIn, out *fuse.AttrOut) syscall.Errno {
	fullPath := rn.path()
	log.S().Infof("Setattr %v, file handle %v", fullPath, fh)

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

	var req = &proto.SetattrReq{Entry: &proto.DirEntry{}}

	if fileHandle, ok := fh.(*proto.FileHandle); ok {
		req.FileHandle = fileHandle
	} else {
		req.Entry.Name = fullPath
	}
	if uid, ok := in.GetUID(); ok {
		rn.uid = uid
		rn.uok = true
		log.S().Debugf("Setattr %v, Uid %v", fullPath, uid)
	}

	if gid, ok := in.GetGID(); ok {
		rn.gid = gid
		rn.gok = true
		log.S().Debugf("Setattr %v, Gid %v", fullPath, gid)
	}

	if mode, ok := in.GetMode(); ok {
		req.Entry.Mode = mode
		req.SetMode = true
		log.S().Debugf("Setattr %v, Mode %v", fullPath, mode)
	}

	if size, ok := in.GetSize(); ok && !rn.ctx.merged {
		req.Entry.Sz = size
		req.SetSize = true
		log.S().Debugf("Setattr %v, Size %v", fullPath, size)
	}
	if req.SetSize {
		cacheFile := rn.cacheFile
		if len(rn.ctx.cacheDir) > 0 && cacheFile != nil && !cacheFile.immutable {
			select {
			case cacheFile.writes <- writeOp{size: int64(req.Entry.Sz)}:
			case <-cacheFile.closeCh:
			}
		}
	}
	var resp *proto.SetattrResp
	var err error
	if req.FileHandle != nil {
		cb := newCallback(&proto.BatchCommandsRequest_Request{Cmd: &proto.BatchCommandsRequest_Request_Setattr{Setattr: req}})
		if err = rn.stream.Send(cb); err != nil {
			log.S().Errorf("Setattr %s send failed: %v", fullPath, err.Error())
			return fs.ToErrno(err)
		}
		res := cb.Wait()
		switch res := res.GetCmd().(type) {
		case *proto.BatchCommandsResponse_Response_Setattr:
			resp = res.Setattr
			if resp.Errno != 0 {
				log.S().Errorf("Setattr %s failed: %v", fullPath, syscall.Errno(resp.Errno))
				return syscall.Errno(resp.Errno)
			}
		}
	} else {
		req.Token = rn.ctx.token
		resp, err = rn.ctx.client.Client().Setattr(ctx1, req)
		if err != nil {
			log.S().Errorf("Setattr %s failed: %v", fullPath, err)
			return fs.ToErrno(err)
		}
	}
	if resp.Errno != 0 {
		log.S().Errorf("Setattr %s failed: %v", fullPath, syscall.Errno(resp.Errno))
		return syscall.Errno(resp.Errno)
	}
	if resp != nil {
		if rn.ctx.merged {
			rn.ctx.mergedfs.RLock()
			entry := rn.ctx.mergedfs.entries[fullPath]
			rn.ctx.mergedfs.RUnlock()
			if entry != nil {
				if entry.Sz < resp.Entry.Sz {
					entry.Sz = resp.Entry.Sz
				}
				if entry.ModTime < resp.Entry.ModTime {
					entry.ModTime = resp.Entry.ModTime
				}
			}
			log.S().Debugf("change %v entry: %v", fullPath, entry)
		}
		rn.e.IsDir = resp.Entry.GetIsDir()
		rn.e.Mode = resp.Entry.GetMode()
		rn.e.Nlink = resp.Entry.GetNlink()
		if rn.ctx.merged {
			rn.e.Ino = rn.inodeHash(fullPath)
		} else {
			rn.e.Ino = resp.Entry.GetIno()
		}
		rn.e.Sz = resp.Entry.GetSz()
		rn.e.ModTime = resp.Entry.GetModTime()
		return rn.Getattr(ctx1, fh, out)
	}
	return syscall.EAGAIN
}

func (rn *RfsNode) Getattr(ctx context.Context, fh fs.FileHandle, out *fuse.AttrOut) syscall.Errno {
	fullPath := rn.path()
	log.S().Debugf("Getattr %v, file handle %v", fullPath, fh)
	if rn.IsRoot() && rn.e.ModTime == 0 {
		entry := &fuse.EntryOut{}
		_, err := rn.Lookup(ctx, "", entry)
		if err != 0 && err != fs.OK {
			return err
		}
		rn.e.Mode = entry.Mode
		rn.e.Nlink = entry.Nlink
		rn.e.Ino = entry.Ino
		rn.e.Sz = entry.Size
		rn.e.ModTime = entry.Mtime*1e9 + uint64(entry.Mtimensec)
		if entry.Uid != uid {
			rn.uid = entry.Uid
			rn.uok = true
		}
		if entry.Gid != gid {
			rn.gid = entry.Gid
			rn.gok = true
		}
	}
	out.Mode = uint32(rn.e.Mode)
	out.Nlink = rn.e.Nlink
	out.Ino = rn.e.Ino
	out.Uid = uid
	if rn.uok {
		out.Uid = rn.uid
	}
	out.Gid = gid
	if rn.gok {
		out.Gid = rn.gid
	}
	out.Size = rn.e.GetSz()
	out.Mtime = rn.e.ModTime / 1e9
	out.Mtimensec = uint32(rn.e.ModTime % 1e9)
	//log.S().Debugf("Getattr %+v, %s ", rn.e, out)
	//if !rn.IsRoot() {
	//	name, p := rn.Parent()
	//	log.S().Debugf("Getattr %v, dir %v, ino %v, parent %v,%v childs %v", fullPath, rn.e.IsDir, rn.e.Ino, name, p, rn.Children())
	//}
	return fs.OK
}

func (rn *RfsNode) Readlink(ctx context.Context) ([]byte, syscall.Errno) {
	if rn.ctx.merged {
		return nil, syscall.ENOSYS
	}
	fullPath := rn.path()
	log.S().Debugf("Readlink %v", fullPath)

	if rn.ctx.merged {
		log.S().Warnf("Readlink %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().Readlink(ctx1, &proto.ReadlinkReq{Token: rn.ctx.token, Name: fullPath})
	if err != nil {
		log.S().Errorf("Readlink %s failed: %v", fullPath, err)
		return nil, fs.ToErrno(err)
	}
	if resp.Errno != 0 {
		log.S().Errorf("Readlink %s failed: %v", fullPath, syscall.Errno(resp.Errno))
		return nil, syscall.Errno(resp.Errno)
	}
	return resp.Buf, fs.OK
}

func (rn *RfsNode) Access(ctx context.Context, mask uint32) syscall.Errno {

	fullPath := rn.path()
	log.S().Debugf("Access %v mask %d", fullPath, mask)

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

	ctx1, cancel := context.WithTimeout(context.Background(), Timeout)
	defer cancel()
	resp, err := rn.ctx.client.Client().Access(ctx1, &proto.AccessReq{Token: rn.ctx.token, Name: fullPath, Mode: mask})
	if err != nil {
		log.S().Errorf("Access %s failed: %v", fullPath, err)
		return fs.ToErrno(err)
	}
	if resp.Errno != 0 {
		log.S().Errorf("Access %s failed: %v", fullPath, syscall.Errno(resp.Errno))
		return syscall.Errno(resp.Errno)
	}
	return fs.OK
}

func (rn *RfsNode) Allocate(ctx context.Context, fh fs.FileHandle, off uint64, size uint64, mode uint32) syscall.Errno {
	fullPath := rn.fullPath
	if len(fullPath) == 0 {
		fullPath = rn.path()
		rn.fullPath = fullPath
	}
	log.S().Debugf("Allocate %v, file handle %v off %d size %d mode %d", fullPath, fh, off, size, mode)

	req := &proto.FallocateReq{FileHandle: getFileHandle(fh), Off: off, Size_: size, Mode: mode}
	cb := newCallback(&proto.BatchCommandsRequest_Request{Cmd: &proto.BatchCommandsRequest_Request_Fallocate{Fallocate: req}})
	if err := rn.stream.Send(cb); err != nil {
		log.S().Errorf("Allocate %s send failed: %v", fullPath, err.Error())
		return syscall.EAGAIN
	}
	res := cb.Wait()
	switch res := res.GetCmd().(type) {
	case *proto.BatchCommandsResponse_Response_Fallocate:
		resp := res.Fallocate
		if resp.Errno != 0 {
			if resp.Errno != int64(syscall.ENOENT) {
				log.S().Errorf("Allocate %s failed: %v", fullPath, syscall.Errno(resp.Errno))
			}
			return syscall.Errno(resp.Errno)
		}
		return fs.OK
	}
	return syscall.EAGAIN
}

const ignoreFile = ".go-fuse."

func AssertFileHandle(fileHandle *proto.FileHandle, fullPath string) {
	if fileHandle != nil {
		if !strings.Contains(fullPath, ignoreFile) {
			logutil.AssertTruef(fileHandle.Name == fullPath, "file name %v mismatch, file handle %v", fullPath, fileHandle)
		}
	}
}