feat: watch vault file changes
This commit is contained in:
@@ -0,0 +1,233 @@
|
||||
// Package filewatcher provides a lightweight live vault change watcher.
|
||||
package filewatcher
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"io/fs"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/verstak/verstak-desktop/internal/core/events"
|
||||
)
|
||||
|
||||
const defaultInterval = 750 * time.Millisecond
|
||||
|
||||
type entryKind string
|
||||
|
||||
const (
|
||||
entryFile entryKind = "file"
|
||||
entryFolder entryKind = "folder"
|
||||
entrySymlink entryKind = "symlink"
|
||||
entryUnknown entryKind = "unknown"
|
||||
)
|
||||
|
||||
type snapshotEntry struct {
|
||||
kind entryKind
|
||||
size int64
|
||||
modTime time.Time
|
||||
}
|
||||
|
||||
// Service polls an open vault and publishes file.changed events for external changes.
|
||||
type Service struct {
|
||||
bus *events.Bus
|
||||
interval time.Duration
|
||||
|
||||
mu sync.Mutex
|
||||
root string
|
||||
cancel chan struct{}
|
||||
done chan struct{}
|
||||
current map[string]snapshotEntry
|
||||
}
|
||||
|
||||
// NewService creates a watcher. The interval parameter is mainly for tests.
|
||||
func NewService(bus *events.Bus, interval time.Duration) *Service {
|
||||
if interval <= 0 {
|
||||
interval = defaultInterval
|
||||
}
|
||||
return &Service{bus: bus, interval: interval}
|
||||
}
|
||||
|
||||
// Start begins watching root. Any previous watch is stopped first.
|
||||
func (s *Service) Start(root string) error {
|
||||
if s == nil {
|
||||
return fmt.Errorf("file watcher is nil")
|
||||
}
|
||||
if root == "" {
|
||||
return fmt.Errorf("file watcher root is empty")
|
||||
}
|
||||
root, err := filepath.Abs(root)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
info, err := os.Stat(root)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if !info.IsDir() {
|
||||
return fmt.Errorf("file watcher root is not a directory: %s", root)
|
||||
}
|
||||
|
||||
initial, err := scan(root)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
s.Stop()
|
||||
|
||||
s.mu.Lock()
|
||||
s.root = root
|
||||
s.current = initial
|
||||
s.cancel = make(chan struct{})
|
||||
s.done = make(chan struct{})
|
||||
cancel := s.cancel
|
||||
done := s.done
|
||||
s.mu.Unlock()
|
||||
|
||||
go s.loop(root, cancel, done)
|
||||
return nil
|
||||
}
|
||||
|
||||
// Stop stops the active watcher.
|
||||
func (s *Service) Stop() {
|
||||
if s == nil {
|
||||
return
|
||||
}
|
||||
s.mu.Lock()
|
||||
cancel := s.cancel
|
||||
done := s.done
|
||||
if cancel == nil {
|
||||
s.mu.Unlock()
|
||||
return
|
||||
}
|
||||
s.cancel = nil
|
||||
s.done = nil
|
||||
s.root = ""
|
||||
s.current = nil
|
||||
close(cancel)
|
||||
s.mu.Unlock()
|
||||
<-done
|
||||
}
|
||||
|
||||
func (s *Service) loop(root string, cancel <-chan struct{}, done chan<- struct{}) {
|
||||
defer close(done)
|
||||
ticker := time.NewTicker(s.interval)
|
||||
defer ticker.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ticker.C:
|
||||
s.poll(root)
|
||||
case <-cancel:
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Service) poll(root string) {
|
||||
next, err := scan(root)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
s.mu.Lock()
|
||||
prev := s.current
|
||||
s.current = next
|
||||
s.mu.Unlock()
|
||||
for path, entry := range next {
|
||||
old, ok := prev[path]
|
||||
if !ok {
|
||||
s.publish(path, "external.create", entry.kind)
|
||||
continue
|
||||
}
|
||||
if entry.kind == entryFile && (entry.size != old.size || !entry.modTime.Equal(old.modTime)) {
|
||||
s.publish(path, "external.update", entry.kind)
|
||||
}
|
||||
}
|
||||
for path, entry := range prev {
|
||||
if _, ok := next[path]; !ok {
|
||||
s.publish(path, "external.delete", entry.kind)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Service) publish(path, operation string, kind entryKind) {
|
||||
if s.bus == nil {
|
||||
return
|
||||
}
|
||||
s.bus.Publish(events.Event{
|
||||
Name: "file.changed",
|
||||
Timestamp: time.Now().UTC().Format(time.RFC3339Nano),
|
||||
Payload: map[string]interface{}{
|
||||
"path": path,
|
||||
"title": path,
|
||||
"operation": operation,
|
||||
"type": string(kind),
|
||||
"workspaceRootPath": workspaceRoot(path),
|
||||
"external": true,
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
func scan(root string) (map[string]snapshotEntry, error) {
|
||||
out := make(map[string]snapshotEntry)
|
||||
err := filepath.WalkDir(root, func(path string, dirEntry fs.DirEntry, err error) error {
|
||||
if err != nil {
|
||||
return nil
|
||||
}
|
||||
if path == root {
|
||||
return nil
|
||||
}
|
||||
rel, err := filepath.Rel(root, path)
|
||||
if err != nil {
|
||||
return nil
|
||||
}
|
||||
rel = filepath.ToSlash(rel)
|
||||
if isReserved(rel) {
|
||||
if dirEntry.IsDir() {
|
||||
return filepath.SkipDir
|
||||
}
|
||||
return nil
|
||||
}
|
||||
info, err := dirEntry.Info()
|
||||
if err != nil {
|
||||
return nil
|
||||
}
|
||||
out[rel] = snapshotEntry{
|
||||
kind: kindFromInfo(info),
|
||||
size: info.Size(),
|
||||
modTime: info.ModTime(),
|
||||
}
|
||||
return nil
|
||||
})
|
||||
return out, err
|
||||
}
|
||||
|
||||
func kindFromInfo(info fs.FileInfo) entryKind {
|
||||
if info.Mode()&os.ModeSymlink != 0 {
|
||||
return entrySymlink
|
||||
}
|
||||
if info.IsDir() {
|
||||
return entryFolder
|
||||
}
|
||||
if info.Mode().IsRegular() {
|
||||
return entryFile
|
||||
}
|
||||
return entryUnknown
|
||||
}
|
||||
|
||||
func isReserved(rel string) bool {
|
||||
first := strings.Split(filepath.ToSlash(rel), "/")[0]
|
||||
return strings.EqualFold(first, ".verstak")
|
||||
}
|
||||
|
||||
func workspaceRoot(path string) string {
|
||||
path = strings.Trim(strings.TrimSpace(filepath.ToSlash(path)), "/")
|
||||
if path == "" {
|
||||
return ""
|
||||
}
|
||||
if idx := strings.Index(path, "/"); idx >= 0 {
|
||||
return path[:idx]
|
||||
}
|
||||
return path
|
||||
}
|
||||
@@ -0,0 +1,86 @@
|
||||
package filewatcher
|
||||
|
||||
import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/verstak/verstak-desktop/internal/core/events"
|
||||
)
|
||||
|
||||
func TestServicePublishesExternalFileChanges(t *testing.T) {
|
||||
root := t.TempDir()
|
||||
bus := events.NewBus()
|
||||
received := make(chan events.Event, 4)
|
||||
bus.Subscribe("file.changed", func(event events.Event) {
|
||||
received <- event
|
||||
})
|
||||
|
||||
service := NewService(bus, 10*time.Millisecond)
|
||||
if err := service.Start(root); err != nil {
|
||||
t.Fatalf("Start: %v", err)
|
||||
}
|
||||
t.Cleanup(service.Stop)
|
||||
|
||||
if err := os.WriteFile(filepath.Join(root, "note.md"), []byte("hello"), 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
event := waitForEvent(t, received)
|
||||
payload, ok := event.Payload.(map[string]interface{})
|
||||
if !ok {
|
||||
t.Fatalf("payload type = %T, want map", event.Payload)
|
||||
}
|
||||
if event.Name != "file.changed" {
|
||||
t.Fatalf("event name = %q, want file.changed", event.Name)
|
||||
}
|
||||
if payload["path"] != "note.md" {
|
||||
t.Fatalf("path = %v, want note.md", payload["path"])
|
||||
}
|
||||
if payload["operation"] != "external.create" {
|
||||
t.Fatalf("operation = %v, want external.create", payload["operation"])
|
||||
}
|
||||
if payload["type"] != "file" {
|
||||
t.Fatalf("type = %v, want file", payload["type"])
|
||||
}
|
||||
}
|
||||
|
||||
func TestServiceIgnoresReservedVerstakPaths(t *testing.T) {
|
||||
root := t.TempDir()
|
||||
if err := os.MkdirAll(filepath.Join(root, ".verstak"), 0o755); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
bus := events.NewBus()
|
||||
received := make(chan events.Event, 4)
|
||||
bus.Subscribe("file.changed", func(event events.Event) {
|
||||
received <- event
|
||||
})
|
||||
|
||||
service := NewService(bus, 10*time.Millisecond)
|
||||
if err := service.Start(root); err != nil {
|
||||
t.Fatalf("Start: %v", err)
|
||||
}
|
||||
t.Cleanup(service.Stop)
|
||||
|
||||
if err := os.WriteFile(filepath.Join(root, ".verstak", "internal.json"), []byte("{}"), 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
select {
|
||||
case event := <-received:
|
||||
t.Fatalf("unexpected reserved-path event: %+v", event)
|
||||
case <-time.After(80 * time.Millisecond):
|
||||
}
|
||||
}
|
||||
|
||||
func waitForEvent(t *testing.T, eventCh <-chan events.Event) events.Event {
|
||||
t.Helper()
|
||||
select {
|
||||
case event := <-eventCh:
|
||||
return event
|
||||
case <-time.After(500 * time.Millisecond):
|
||||
t.Fatal("timed out waiting for file.changed")
|
||||
return events.Event{}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user