Use near-real-time change notifications for fetching delta

cmd/onedriver/main.go

@@ -1,13 +1,11 @@
 package main
 
 import (
-	"encoding/json"
 	"fmt"
 	"io/ioutil"
 	"os"
 	"os/signal"
 	"path/filepath"
-	"strings"
 	"syscall"
 	"time"
 
@@ -60,6 +58,10 @@ func main() {
 	versionFlag := flag.BoolP("version", "v", false, "Display program version.")
 	debugOn := flag.BoolP("debug", "d", false, "Enable FUSE debug logging. "+
 		"This logs communication between onedriver and the kernel.")
+	allowOther := flag.BoolP("allow-other", "", false,
+		"Allow access to the mount point for other users.")
+	uid := flag.Uint32P("uid", "", uint32(os.Getuid()), "Owner uid of the mount point.")
+	gid := flag.Uint32P("gid", "", uint32(os.Getgid()), "Owner gid of the mount point.")
 	help := flag.BoolP("help", "h", false, "Displays this help message.")
 	flag.Usage = usage
 	flag.Parse()
@@ -125,13 +127,15 @@ func main() {
 	// create the filesystem
 	log.Info().Msgf("onedriver %s", common.Version())
 	auth := graph.Authenticate(config.AuthConfig, authPath, *headless)
-	filesystem := fs.NewFilesystem(auth, cachePath)
-	go filesystem.DeltaLoop(30 * time.Second)
+	filesystem := fs.NewFilesystem(auth, cachePath, fs.OptionOwner(*uid, *gid))
+	go filesystem.DeltaLoop(10 * time.Minute)
 	xdgVolumeInfo(filesystem, auth)
 
 	server, err := fuse.NewServer(filesystem, mountpoint, &fuse.MountOptions{
 		Name:          "onedriver",
 		FsName:        "onedriver",
+		DirectMount:   true,
+		AllowOther:    *allowOther,
 		DisableXAttrs: true,
 		MaxBackground: 1024,
 		Debug:         *debugOn,
@@ -153,34 +157,3 @@ func main() {
 		Msg("Serving filesystem.")
 	server.Serve()
 }
-
-// xdgVolumeInfo createx .xdg-volume-info for a nice little onedrive logo in the
-// corner of the mountpoint and shows the account name in the nautilus sidebar
-func xdgVolumeInfo(filesystem *fs.Filesystem, auth *graph.Auth) {
-	if child, _ := filesystem.GetPath("/.xdg-volume-info", auth); child != nil {
-		return
-	}
-	log.Info().Msg("Creating .xdg-volume-info")
-	user, err := graph.GetUser(auth)
-	if err != nil {
-		log.Error().Err(err).Msg("Could not create .xdg-volume-info")
-		return
-	}
-	xdgVolumeInfo := common.TemplateXDGVolumeInfo(user.UserPrincipalName)
-
-	// just upload directly and shove it in the cache
-	// (since the fs isn't mounted yet)
-	resp, err := graph.Put(
-		graph.ResourcePath("/.xdg-volume-info")+":/content",
-		auth,
-		strings.NewReader(xdgVolumeInfo),
-	)
-	if err != nil {
-		log.Error().Err(err).Msg("Failed to write .xdg-volume-info")
-	}
-	root, _ := filesystem.GetPath("/", auth) // cannot fail
-	inode := fs.NewInode(".xdg-volume-info", 0644, root)
-	if json.Unmarshal(resp, &inode) == nil {
-		filesystem.InsertID(inode.ID(), inode)
-	}
-}

cmd/onedriver/xdg.go

@@ -0,0 +1,45 @@
+//go:build linux && cgo
+// +build linux,cgo
+
+package main
+
+import (
+	"encoding/json"
+	"strings"
+
+	"github.com/jstaf/onedriver/fs"
+	"github.com/jstaf/onedriver/cmd/common"
+	"github.com/jstaf/onedriver/fs/graph"
+	"github.com/rs/zerolog/log"
+)
+
+// xdgVolumeInfo createx .xdg-volume-info for a nice little onedrive logo in the
+// corner of the mountpoint and shows the account name in the nautilus sidebar
+func xdgVolumeInfo(filesystem *fs.Filesystem, auth *graph.Auth) {
+	if child, _ := filesystem.GetPath("/.xdg-volume-info", auth); child != nil {
+		return
+	}
+	log.Info().Msg("Creating .xdg-volume-info")
+	user, err := graph.GetUser(auth)
+	if err != nil {
+		log.Error().Err(err).Msg("Could not create .xdg-volume-info")
+		return
+	}
+	xdgVolumeInfo := common.TemplateXDGVolumeInfo(user.UserPrincipalName)
+
+	// just upload directly and shove it in the cache
+	// (since the fs isn't mounted yet)
+	resp, err := graph.Put(
+		graph.ResourcePath("/.xdg-volume-info")+":/content",
+		auth,
+		strings.NewReader(xdgVolumeInfo),
+	)
+	if err != nil {
+		log.Error().Err(err).Msg("Failed to write .xdg-volume-info")
+	}
+	root, _ := filesystem.GetPath("/", auth) // cannot fail
+	inode := fs.NewInode(".xdg-volume-info", 0644, root)
+	if json.Unmarshal(resp, &inode) == nil {
+		filesystem.InsertID(inode.ID(), inode)
+	}
+}

cmd/onedriver/xdg_headless.go

@@ -0,0 +1,13 @@
+//go:build !linux || !cgo
+// +build !linux !cgo
+
+package main
+
+import (
+	"github.com/jstaf/onedriver/fs"
+	"github.com/jstaf/onedriver/fs/graph"
+)
+
+func xdgVolumeInfo(filesystem *fs.Filesystem, auth *graph.Auth) {
+	return
+}

fs/cache.go

@@ -21,13 +21,17 @@ import (
 type Filesystem struct {
 	fuse.RawFileSystem
 
-	metadata  sync.Map
-	db        *bolt.DB
-	content   *LoopbackCache
-	auth      *graph.Auth
-	root      string // the id of the filesystem's root item
-	deltaLink string
-	uploads   *UploadManager
+	uid uint32
+	gid uint32
+
+	metadata             sync.Map
+	db                   *bolt.DB
+	content              *LoopbackCache
+	auth                 *graph.Auth
+	root                 string // the id of the filesystem's root item
+	deltaLink            string
+	subscribeChangesLink string
+	uploads              *UploadManager
 
 	sync.RWMutex
 	offline    bool
@@ -50,8 +54,17 @@ var (
 // so we can tell what format the db has
 const fsVersion = "1"
 
+type Option func(fs *Filesystem)
+
+func OptionOwner(uid, gid uint32) Option {
+	return func(fs *Filesystem) {
+		fs.uid = uid
+		fs.gid = gid
+	}
+}
+
 // NewFilesystem creates a new filesystem
-func NewFilesystem(auth *graph.Auth, cacheDir string) *Filesystem {
+func NewFilesystem(auth *graph.Auth, cacheDir string, opts ...Option) *Filesystem {
 	// prepare cache directory
 	if _, err := os.Stat(cacheDir); err != nil {
 		if err = os.Mkdir(cacheDir, 0700); err != nil {
@@ -102,12 +115,19 @@ func NewFilesystem(auth *graph.Auth, cacheDir string) *Filesystem {
 
 	// ok, ready to start fs
 	fs := &Filesystem{
+		// default: whatever user is running the filesystem is the owner
+		uid: uint32(os.Getuid()),
+		gid: uint32(os.Getgid()),
+
 		RawFileSystem: fuse.NewDefaultRawFileSystem(),
 		content:       content,
 		db:            db,
 		auth:          auth,
 		opendirs:      make(map[uint64][]*Inode),
 	}
+	for _, opt := range opts {
+		opt(fs)
+	}
 
 	rootItem, err := graph.GetItem("root", auth)
 	root := NewInodeDriveItem(rootItem)
@@ -164,6 +184,7 @@ func NewFilesystem(auth *graph.Auth, cacheDir string) *Filesystem {
 		// using token=latest because we don't care about existing items - they'll
 		// be downloaded on-demand by the cache
 		fs.deltaLink = "/me/drive/root/delta?token=latest"
+		fs.subscribeChangesLink = "/me/drive/root/subscriptions/socketIo"
 	}
 
 	// deltaloop is started manually

fs/content_cache.go

@@ -100,10 +100,9 @@ func (l *LoopbackCache) Open(id string) (*os.File, error) {
 
 // Close closes the currently open fd
 func (l *LoopbackCache) Close(id string) {
-	if fd, ok := l.fds.Load(id); ok {
+	if fd, ok := l.fds.LoadAndDelete(id); ok {
 		file := fd.(*os.File)
 		file.Sync()
 		file.Close()
-		l.fds.Delete(id)
 	}
 }

fs/delta.go

@@ -15,6 +15,11 @@ import (
 // called as a goroutine
 func (f *Filesystem) DeltaLoop(interval time.Duration) {
 	log.Trace().Msg("Starting delta goroutine.")
+
+	subsc := newSubscription(f.subscribeChanges)
+	go subsc.Start()
+	defer subsc.Stop()
+
 	for { // eva
 		// get deltas
 		log.Trace().Msg("Fetching deltas from server.")
@@ -63,6 +68,7 @@ func (f *Filesystem) DeltaLoop(interval time.Duration) {
 			f.SerializeAll()
 		}
 
+		waitDur := interval
 		if pollSuccess {
 			f.Lock()
 			if f.offline {
@@ -76,10 +82,14 @@ func (f *Filesystem) DeltaLoop(interval time.Duration) {
 			})
 
 			// wait until next interval
-			time.Sleep(interval)
 		} else {
 			// shortened duration while offline
-			time.Sleep(2 * time.Second)
+			waitDur = 2 * time.Second
+		}
+
+		select {
+		case <-time.After(waitDur):
+		case <-subsc.C:
 		}
 	}
 }

fs/fs.go

@@ -115,6 +115,25 @@ func isNameRestricted(name string) bool {
 	return disallowedRexp.FindStringIndex(name) != nil
 }
 
+// makeattr is a convenience function to create a set of filesystem attrs for
+// use with syscalls that use or modify attrs.
+func (f *Filesystem) makeAttr(i *Inode) fuse.Attr {
+	mtime := i.ModTime()
+	return fuse.Attr{
+		Ino:   i.NodeID(),
+		Size:  i.Size(),
+		Nlink: i.NLink(),
+		Ctime: mtime,
+		Mtime: mtime,
+		Atime: mtime,
+		Mode:  i.Mode(),
+		Owner: fuse.Owner{
+			Uid: f.uid,
+			Gid: f.gid,
+		},
+	}
+}
+
 // Statfs returns information about the filesystem. Mainly useful for checking
 // quotas and storage limits.
 func (f *Filesystem) StatFs(cancel <-chan struct{}, in *fuse.InHeader, out *fuse.StatfsOut) fuse.Status {
@@ -180,7 +199,7 @@ func (f *Filesystem) Mkdir(cancel <-chan struct{}, in *fuse.MkdirIn, name string
 	newInode.mode = in.Mode | fuse.S_IFDIR
 
 	out.NodeId = f.InsertChild(id, newInode)
-	out.Attr = newInode.makeAttr()
+	out.Attr = f.makeAttr(newInode)
 	out.SetAttrTimeout(timeout)
 	out.SetEntryTimeout(timeout)
 	return fuse.OK
@@ -305,7 +324,7 @@ func (f *Filesystem) ReadDirPlus(cancel <-chan struct{}, in *fuse.ReadIn, out *f
 		return fuse.EIO
 	}
 	entryOut.NodeId = entry.Ino
-	entryOut.Attr = inode.makeAttr()
+	entryOut.Attr = f.makeAttr(inode)
 	entryOut.SetAttrTimeout(timeout)
 	entryOut.SetEntryTimeout(timeout)
 	return fuse.OK
@@ -369,7 +388,7 @@ func (f *Filesystem) Lookup(cancel <-chan struct{}, in *fuse.InHeader, name stri
 	}
 
 	out.NodeId = child.NodeID()
-	out.Attr = child.makeAttr()
+	out.Attr = f.makeAttr(child)
 	out.SetAttrTimeout(timeout)
 	out.SetEntryTimeout(timeout)
 	return fuse.OK
@@ -412,7 +431,7 @@ func (f *Filesystem) Mknod(cancel <-chan struct{}, in *fuse.MknodIn, name string
 		Str("mode", Octal(in.Mode)).
 		Msg("Creating inode.")
 	out.NodeId = f.InsertChild(parentID, inode)
-	out.Attr = inode.makeAttr()
+	out.Attr = f.makeAttr(inode)
 	out.SetAttrTimeout(timeout)
 	out.SetEntryTimeout(timeout)
 	return fuse.OK
@@ -721,7 +740,7 @@ func (f *Filesystem) GetAttr(cancel <-chan struct{}, in *fuse.GetAttrIn, out *fu
 		Str("path", inode.Path()).
 		Msg("")
 
-	out.Attr = inode.makeAttr()
+	out.Attr = f.makeAttr(inode)
 	out.SetTimeout(timeout)
 	return fuse.OK
 }
@@ -784,7 +803,7 @@ func (f *Filesystem) SetAttr(cancel <-chan struct{}, in *fuse.SetAttrIn, out *fu
 	}
 
 	i.Unlock()
-	out.Attr = i.makeAttr()
+	out.Attr = f.makeAttr(i)
 	out.SetTimeout(timeout)
 	return fuse.OK
 }

fs/inode.go

@@ -3,7 +3,6 @@ package fs
 import (
 	"encoding/json"
 	"math/rand"
-	"os"
 	"strconv"
 	"strings"
 	"sync"
@@ -211,26 +210,6 @@ func (i *Inode) HasChildren() bool {
 	return len(i.children) > 0
 }
 
-// makeattr is a convenience function to create a set of filesystem attrs for
-// use with syscalls that use or modify attrs.
-func (i *Inode) makeAttr() fuse.Attr {
-	mtime := i.ModTime()
-	return fuse.Attr{
-		Ino:   i.NodeID(),
-		Size:  i.Size(),
-		Nlink: i.NLink(),
-		Ctime: mtime,
-		Mtime: mtime,
-		Atime: mtime,
-		Mode:  i.Mode(),
-		// whatever user is running the filesystem is the owner
-		Owner: fuse.Owner{
-			Uid: uint32(os.Getuid()),
-			Gid: uint32(os.Getgid()),
-		},
-	}
-}
-
 // IsDir returns if it is a directory (true) or file (false).
 func (i *Inode) IsDir() bool {
 	// 0 if the dir bit is not set