diff --git a/client/go.mod b/client/go.mod index e515a8b..0aaba1e 100644 --- a/client/go.mod +++ b/client/go.mod @@ -11,6 +11,7 @@ require ( friendnet.org/updater v0.0.0 friendnet.org/upnp v0.0.0 friendnet.org/webui v0.0.0 + github.com/fsnotify/fsnotify v1.10.1 github.com/google/uuid v1.6.0 github.com/pkg/browser v0.0.0-20240102092130-5ac0b6a4141c github.com/quic-go/quic-go v0.61.0 @@ -22,7 +23,6 @@ require ( require ( github.com/dustin/go-humanize v1.0.1 // indirect - github.com/fsnotify/fsnotify v1.10.1 // indirect github.com/mattn/go-isatty v0.0.20 // indirect github.com/ncruces/go-strftime v1.0.0 // indirect github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect diff --git a/client/go.sum b/client/go.sum index c6be927..1407aa5 100644 --- a/client/go.sum +++ b/client/go.sum @@ -23,8 +23,9 @@ github.com/pkg/browser v0.0.0-20240102092130-5ac0b6a4141c h1:+mdjkGKdHQG3305AYmd github.com/pkg/browser v0.0.0-20240102092130-5ac0b6a4141c/go.mod h1:7rwL4CYBLnjLxUqIJNnCWiEdr3bn6IUYi15bNlnbCCU= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= -github.com/quic-go/quic-go v0.59.0 h1:OLJkp1Mlm/aS7dpKgTc6cnpynnD2Xg7C1pwL6vy/SAw= -github.com/quic-go/quic-go v0.59.0/go.mod h1:upnsH4Ju1YkqpLXC305eW3yDZ4NfnNbmQRCMWS58IKU= +github.com/quic-go/go-ossfuzz-seeds v0.1.0 h1:APacT+iIaNF6fd8AGEiN3bT/Jtkd2jz4v4TzM7MFjy0= +github.com/quic-go/go-ossfuzz-seeds v0.1.0/go.mod h1:3IOHRbJIc+L6YKMwfDtJAM9Vj9k0YY4muhuyUYk5tbk= +github.com/quic-go/quic-go v0.61.0 h1:ui88A53s8MSVYLC56en0KQ17HARk+9986Dn0SBfKNvA= github.com/quic-go/quic-go v0.61.0/go.mod h1:9So2anK4Tp22URSQq00k+Vo2PNkle96ycDPDHL4s9vs= github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE= github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo= @@ -32,29 +33,24 @@ github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= go.uber.org/mock v0.5.2 h1:LbtPTcP8A5k9WPXj54PPPbjcI4Y6lhyOZXn+VS7wNko= go.uber.org/mock v0.5.2/go.mod h1:wLlUxC2vVTPTaE3UD51E0BGOAElKrILxhVSDYQLld5o= -golang.org/x/crypto v0.48.0 h1:/VRzVqiRSggnhY7gNRxPauEQ5Drw9haKdM0jqfcCFts= -golang.org/x/crypto v0.48.0/go.mod h1:r0kV5h3qnFPlQnBSrULhlsRfryS2pmewsg+XfMgkVos= +golang.org/x/crypto v0.54.0 h1:YLIA59K4fiNzHzjnZt2tUJQjQtUWfWbeHBqKtk3eScw= golang.org/x/crypto v0.54.0/go.mod h1:KWL8ny2AZdGR2cWmzeHrp2azQPGogOv+HeQaVEXC2dk= golang.org/x/exp v0.0.0-20260218203240-3dfff04db8fa h1:Zt3DZoOFFYkKhDT3v7Lm9FDMEV06GpzjG2jrqW+QTE0= golang.org/x/exp v0.0.0-20260218203240-3dfff04db8fa/go.mod h1:K79w1Vqn7PoiZn+TkNpx3BUWUQksGO3JcVX6qIjytmA= -golang.org/x/mod v0.33.0 h1:tHFzIWbBifEmbwtGz65eaWyGiGZatSrT9prnU8DbVL8= -golang.org/x/mod v0.33.0/go.mod h1:swjeQEj+6r7fODbD2cqrnje9PnziFuw4bmLbBZFrQ5w= -golang.org/x/net v0.50.0 h1:ucWh9eiCGyDR3vtzso0WMQinm2Dnt8cFMuQa9K33J60= -golang.org/x/net v0.50.0/go.mod h1:UgoSli3F/pBgdJBHCTc+tp3gmrU4XswgGRgtnwWTfyM= +golang.org/x/mod v0.37.0 h1:vF1DjpVEshcIqoEaauuHebaLk1O1forxjxBaVn884JQ= +golang.org/x/mod v0.37.0/go.mod h1:m8S8VeM9r4dzDwjrKO0a1sZP3YjeMamRRlD+fmR2Q/0= +golang.org/x/net v0.56.0 h1:Rw8j/hFzGvJUZwNBXnAtf5sVDVt+65SK2C7IxCxZt5o= golang.org/x/net v0.56.0/go.mod h1:D3Ku6r+V6JROoZK144D2XfMHFcMq/0zSfLelVTCFKec= -golang.org/x/sync v0.19.0 h1:vV+1eWNmZ5geRlYjzm2adRgW2/mcpevXNg50YZtPCE4= -golang.org/x/sync v0.19.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI= +golang.org/x/sync v0.22.0 h1:SZjpbeLmrCk4xhRSZFNZW5gFUeCeFgjekvI/+gfScek= +golang.org/x/sync v0.22.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= golang.org/x/sys v0.1.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/sys v0.41.0 h1:Ivj+2Cp/ylzLiEU89QhWblYnOE9zerudt9Ftecq2C6k= -golang.org/x/sys v0.41.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks= golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs= golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= -golang.org/x/text v0.34.0 h1:oL/Qq0Kdaqxa1KbNeMKwQq0reLCCaFtqu2eNuSeNHbk= -golang.org/x/text v0.34.0/go.mod h1:homfLqTYRFyVYemLBFl5GgL/DWEiH5wcsQ5gSh1yziA= +golang.org/x/text v0.40.0 h1:Ub2Z6/xjgF1WrYQz2nuITOEegKFtiIy+rieRJ5lHZKs= golang.org/x/text v0.40.0/go.mod h1:hpnzDAfGV753zIKo+wk3u1bVKCGPbrnF7+7LBF/UHVY= -golang.org/x/tools v0.42.0 h1:uNgphsn75Tdz5Ji2q36v/nsFSfR/9BRFvqhGBaJGd5k= -golang.org/x/tools v0.42.0/go.mod h1:Ma6lCIwGZvHK6XtgbswSoWroEkhugApmsXyrUmBhfr0= +golang.org/x/tools v0.47.0 h1:7Kn5x/d1svx/PzryTsqeoZN4TZwqeH5pGWjefhLi/1Q= +golang.org/x/tools v0.47.0/go.mod h1:dFHnyTvFWY212G+h7ZY4Vsp/K3U4/7W9TyVaAul8uCA= google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE= google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= diff --git a/client/share/manager.go b/client/share/manager.go index 1ace28c..4e35986 100644 --- a/client/share/manager.go +++ b/client/share/manager.go @@ -31,6 +31,9 @@ var ErrTooManyFiles = errors.New("too many files in share, indexing canceled") // ErrInvalidShareName is returned when trying to create a share with an invalid name. var ErrInvalidShareName = errors.New("invalid share name") +// ErrWrongShare is returned when the wrong share is used in reference to an operation involving indices. +var ErrWrongShare = errors.New("wrong share referenced in index operation") + type shareData struct { share Share record storage.ShareRecord @@ -75,20 +78,6 @@ func NewManager( return nil, fmt.Errorf(`failed to get share records for server %q: %w`, serverUuid, err) } - shareMap := make(map[string]*shareData, len(records)) - for _, record := range records { - var share Share - share, err = NewDirShare( - record.Name, - record.Path.String(), - record.FollowLinks, - ) - shareMap[record.Name] = &shareData{ - share: share, - record: record, - } - } - m := &Manager{ ctx: ctx, ctxCancel: ctxCancel, @@ -98,14 +87,33 @@ func NewManager( serverUuid: serverUuid, storage: storage, - shareMap: shareMap, - indexerInterval: 1 * time.Hour, indexingShares: make(map[string]struct{}), indexerMaxFiles: 1_000_000, orphanedIndexGcInterval: 10 * time.Minute, } + m.shareMap = make(map[string]*shareData, len(records)) + for _, record := range records { + var share Share + share, err = NewDirShare( + ctx, + record.Name, + record.Path.String(), + record.FollowLinks, + ) + + if record.EnableIndexing && share.SupportsWatching() { + share.OnNeedIndex(m.buildIndexCallback(ctx, record.Name)) + share.OnDelete(m.buildDeleteCallback(ctx, record.Uuid, record.Path)) + } + + m.shareMap[record.Name] = &shareData{ + share: share, + record: record, + } + } + go m.indexerDaemon() go m.orphanedIndexGc() @@ -204,6 +212,101 @@ func (m *Manager) orphanedIndexGc() { } } +// indexShareFile will index a single file in the share with the specified name. +// This function is not optimized for bulk indexing. In that case, use indexShare. +// This function should only be called from file watcher callbacks (see share.go). +// This function expects the provided path to point to a file. +// Refuses to index the file if it resides outside of the given share, returning ErrWrongShare. +// Refuses to index the file if the share has indexing disabled, returning ErrIndexingDisabled. +func (m *Manager) indexShareFile(ctx context.Context, name string, path common.ProtoPath) error { + if path.IsZero() || path.IsRoot() { + return nil + } + + m.mu.Lock() + val, has := m.shareMap[name] + if !has { + m.mu.Unlock() + return nil + } + + pathContains, pathIdx := common.PathContains(val.record.Path, path) + if !pathContains { + m.mu.Unlock() + return ErrWrongShare + } + + relPath, err := common.SegmentsToPath(path.ToSegments()[pathIdx:]) + if err != nil { + m.mu.Unlock() + return err + } + + curIndexId := time.Now().UnixMilli() + val.lastIndexId = curIndexId + m.mu.Unlock() + + share := val.share + rec := val.record + + if !rec.EnableIndexing { + return ErrIndexingDisabled + } + + meta, err := share.GetFileMeta(relPath) + if err != nil { + return err + } + + // TODO Consider doing garbage collection after this + // TODO Figure out symlinks + + err = m.storage.InsertShareIndex( + ctx, + rec.Uuid, + curIndexId, + relPath.String(), + false, + int64(meta.GetSize()), + ) + if err != nil { + return err + } + + m.logger.Debug("indexed share index", "service", "share.Manager", "path", relPath.String()) + + return nil +} + +func (m *Manager) buildIndexCallback(ctx context.Context, name string) func(path common.ProtoPath) { + return func(path common.ProtoPath) { + _ = m.indexShareFile(ctx, name, path) + } +} + +func (m *Manager) buildDeleteCallback(ctx context.Context, uuid string, parent common.ProtoPath) func(path common.ProtoPath) { + return func(path common.ProtoPath) { + if path.IsZero() || path.IsRoot() { + return + } + + pathContains, pathIdx := common.PathContains(parent, path) + if !pathContains { + return + } + + relPath, err := common.SegmentsToPath(path.ToSegments()[pathIdx:]) + if err != nil { + return + } + if relPath.IsZero() || relPath.IsRoot() { + return + } + + _ = m.storage.DeleteShareIndexByPath(ctx, uuid, relPath.String()) + } +} + // indexShare indexes all files in the share with the specified name. // It returns the number of files indexed, whether the share existed, and any error that occurred. // Refuses to index the share if it has indexing disabled, returning ErrIndexingDisabled. @@ -478,6 +581,7 @@ func (m *Manager) Add( // Create instance. share, err := NewDirShare( + ctx, name, path, followLinks, @@ -494,6 +598,12 @@ func (m *Manager) Add( m.mu.Unlock() if rec.EnableIndexing { + // Add requisite callbacks for file watcher + if share.SupportsWatching() { + share.OnNeedIndex(m.buildIndexCallback(ctx, name)) + share.OnDelete(m.buildDeleteCallback(ctx, rec.Uuid, rec.Path)) + } + go func() { m.indexShareWithLockAndLogging(rec) }() diff --git a/client/share/share.go b/client/share/share.go index 5b837d7..378b040 100644 --- a/client/share/share.go +++ b/client/share/share.go @@ -1,20 +1,25 @@ package share import ( + "context" "errors" "io" "io/fs" "os" "path/filepath" "strings" + "sync" "friendnet.org/common" pb "friendnet.org/protocol/pb/v1" + "github.com/fsnotify/fsnotify" ) // ErrShareClosed is returned by Share methods when the share is closed. var ErrShareClosed = errors.New("share closed") +type ShareCallback func(path common.ProtoPath) + // Share is a shared filesystem. // A share only has the concepts of files and directories. // It has no way of representing symlinks or pipes. @@ -60,26 +65,47 @@ type Share interface { // // May return ErrShareClosed if the share is closed, depending on the implementation. GetFile(path common.ProtoPath, offset uint64, limit uint64) (*pb.MsgFileMeta, io.ReadCloser, error) + + // SupportsWatching will return true if the Share implementation supports filesystem event watching + SupportsWatching() bool + + // OnNeedIndex subscribes a callback to a filesystem event listener. + // The callbacks will fire, in order of subscription, when a new file in a watched directory is created or if an existing file has been modified. + OnNeedIndex(callback ShareCallback) + + // OnDelete subscribes a callback to a filesystem event listener. + // The callbacks will fire, in order of subscription, when a file in a watched directory is deleted. + OnDelete(callback ShareCallback) } // DirShare is an implementation of Share backed by a directory. type DirShare struct { + ctx context.Context + name string dir string followLinks bool fsys fs.FS + + // Watching related members + mu sync.RWMutex + + watcher *fsnotify.Watcher + onIndexHdlrs []ShareCallback + onDeleteHdlrs []ShareCallback } var _ Share = (*DirShare)(nil) -// Close is no-op because DirShare is stateless. func (s *DirShare) Close() error { - return nil + return s.watcher.Close() } // NewDirShare creates a new DirShare backed by the specified directory. +// It will also initialize a filesystem watcher. // If followLinks is false, symlinks will be treated as if they do not exist. func NewDirShare( + ctx context.Context, name string, dir string, followLinks bool, @@ -89,12 +115,88 @@ func NewDirShare( return nil, err } - return &DirShare{ + // Setup watcher + watcher, err := fsnotify.NewWatcher() + if err != nil { + return nil, err + } + + share := &DirShare{ + ctx: ctx, name: name, dir: abs, followLinks: followLinks, fsys: os.DirFS(abs), - }, nil + watcher: watcher, + } + + err = watcher.Add(abs) + if err != nil { + return nil, err + } + + // Init watcher + // On errors, just kill the watcher + go func() { + for { + select { + case <-ctx.Done(): + return + case event, ok := <-watcher.Events: + if !ok { + break + } + + share.mu.RLock() + + if len(event.Name) == 0 { + continue + } + + // TODO Is this the best way to reject a directory + if event.Name[len(event.Name)-1] == '/' { + continue + } + + path, err := common.NormalizePath(event.Name) + if err != nil { + break + } + + if event.Has(fsnotify.Create) || event.Has(fsnotify.Write) || event.Has(fsnotify.Chmod) { + for _, cb := range share.onIndexHdlrs { + cb(path) + } + } else if event.Has(fsnotify.Remove) { + for _, cb := range share.onDeleteHdlrs { + cb(path) + } + } + + share.mu.RUnlock() + } + } + }() + + return share, nil +} + +func (s *DirShare) SupportsWatching() bool { + return true +} + +func (s *DirShare) OnNeedIndex(callback ShareCallback) { + s.mu.Lock() + defer s.mu.Unlock() + + s.onIndexHdlrs = append(s.onIndexHdlrs, callback) +} + +func (s *DirShare) OnDelete(callback ShareCallback) { + s.mu.Lock() + defer s.mu.Unlock() + + s.onDeleteHdlrs = append(s.onDeleteHdlrs, callback) } func (s *DirShare) isInfoOk(info fs.FileInfo) bool { diff --git a/client/storage/storage.go b/client/storage/storage.go index e24de2f..12b9a74 100644 --- a/client/storage/storage.go +++ b/client/storage/storage.go @@ -298,6 +298,14 @@ func (s *Storage) DeleteShareByUuid( return err } +func (s *Storage) DeleteShareIndexByPath(ctx context.Context, uuid string, path string) error { + _, err := s.Exec(ctx, `delete from share_index_fts where share = ? and path = ?`, uuid, path) + if err != nil { + return fmt.Errorf("failed to delete index for %q in share %q: %w", path, uuid, err) + } + return nil +} + // ClearShareIndex clears the search index for the share with the specified UUID. // It excludes all indexes that have an index ID lower than curIndexId. func (s *Storage) ClearShareIndex(ctx context.Context, uuid string, curIndexId int64) error { diff --git a/common/path.go b/common/path.go index 932e050..a21fce1 100644 --- a/common/path.go +++ b/common/path.go @@ -262,3 +262,30 @@ func JoinPaths(paths ...ProtoPath) ProtoPath { } return UncheckedCreateProtoPath(sb.String()) } + +// PathContains returns true when path is contained wholly in parent and the array index in the path array corresponding to the full parent path. +func PathContains(parent ProtoPath, path ProtoPath) (bool, int) { + if parent.IsZero() || path.IsZero() { + return false, 0 + } + + if parent.IsRoot() { + return true, 0 + } + + parentDiv := parent.ToSegments() + pathDiv := path.ToSegments() + + if len(parentDiv) > len(pathDiv) { + return false, 0 + } + + i := 0 + for ; i < len(parentDiv); i++ { + if parentDiv[i] != pathDiv[i] { + return false, i + } + } + + return true, i +} diff --git a/common/path_test.go b/common/path_test.go index b9d7a45..3fc00b2 100644 --- a/common/path_test.go +++ b/common/path_test.go @@ -2,6 +2,91 @@ package common import "testing" +//goland:noinspection ALL +func TestPathContains(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + parent string + path string + wantBool bool + wantIdx int + }{ + // --- valid --- + { + name: "valid_root_whatever", + parent: "/", + path: "/this/is/anything", + wantBool: true, + wantIdx: 0, + }, + { + name: "valid_first_degree", + parent: "/foo/", + path: "/foo/bar", + wantBool: true, + wantIdx: 1, + }, + { + name: "valid_second_degree", + parent: "/foo/", + path: "/foo/bar/qux", + wantBool: true, + wantIdx: 1, + }, + // --- invalid --- + { + name: "invalid_longer_parent", + parent: "/foo/bar/qux", + path: "/foo/", + wantBool: false, + wantIdx: 0, + }, + { + name: "invalid_wrong_first_degree", + parent: "/qux/", + path: "/foo/bar/qux", + wantBool: false, + wantIdx: 0, + }, + { + name: "invalid_wrong_second_degree", + parent: "/qux/foo", + path: "/foo/bar/qux", + wantBool: false, + wantIdx: 0, + }, + } + + for _, tt := range tests { + tt := tt + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + parent, err := NormalizePath(tt.parent) + if err != nil { + t.Fatalf("PathContains parent normalize fail: %v", err) + } + + path, err := NormalizePath(tt.path) + if err != nil { + t.Fatalf("PathContains path normalize fail: %v", err) + } + + res, idx := PathContains(parent, path) + + if res != tt.wantBool { + t.Fatalf("expected result %t, got %t", tt.wantBool, res) + } + + if idx != tt.wantIdx { + t.Fatalf("expected parent path cutoff index %d, got %d", tt.wantIdx, idx) + } + }) + } +} + //goland:noinspection ALL func TestValidatePath(t *testing.T) { t.Parallel()