feature (watch): improve watch api

server/pkg/files/handler_mkdir.go

@@ -1,7 +1,6 @@
 package files
 
 import (
-	"strings"
 	"net/http"
 
 	. "github.com/mickael-kerjean/filestash/server/pkg/core"
@@ -39,5 +38,5 @@ func FileMkdir(ctx *App, res http.ResponseWriter, req *http.Request) {
 		return
 	}
 	SendSuccessResult(res, nil)
-	markDirty(ctx, req, strings.TrimSuffix(path, "/"))
+	EventWatch(ctx, req, "mkdir", path)
 }

server/pkg/files/handler_mv.go

@@ -49,6 +49,5 @@ func FileMv(ctx *App, res http.ResponseWriter, req *http.Request) {
 		return
 	}
 	SendSuccessResult(res, nil)
-	markDirty(ctx, req, from)
-	markDirty(ctx, req, to)
+	EventWatch(ctx, req, "mv", from, to)
 }

server/pkg/files/handler_rm.go

@@ -38,5 +38,5 @@ func FileRm(ctx *App, res http.ResponseWriter, req *http.Request) {
 		return
 	}
 	SendSuccessResult(res, nil)
-	markDirty(ctx, req, path)
+	EventWatch(ctx, req, "rm", path)
 }

server/pkg/files/handler_save.go

@@ -88,7 +88,7 @@ func FileSave(ctx *App, res http.ResponseWriter, req *http.Request) {
 		SendErrorResult(res, ErrNotImplemented)
 		return
 	}
-	markDirty(ctx, req, path)
+	EventWatch(ctx, req, "save", path)
 }
 
 func handlerClassic(ctx *App, res http.ResponseWriter, req *http.Request, path string, h http.Header) {

server/pkg/files/handler_touch.go

@@ -38,5 +38,5 @@ func FileTouch(ctx *App, res http.ResponseWriter, req *http.Request) {
 		return
 	}
 	SendSuccessResult(res, nil)
-	markDirty(ctx, req, path)
+	EventWatch(ctx, req, "touch", path)
 }

server/pkg/files/handler_watch.go

@@ -2,12 +2,11 @@ package files
 
 import (
 	"container/ring"
-	"strings"
 	"encoding/json"
 	"fmt"
 	"net/http"
+	"strings"
 	"sync"
-	"sync/atomic"
 	"time"
 
 	. "github.com/mickael-kerjean/filestash/server/pkg/core"
@@ -16,22 +15,23 @@ import (
 )
 
 var (
-	feed = ring.New(1000)
-	flock = sync.Mutex{}
-	fdrop = atomic.Uint64{}
+	journal = ring.New(5000)
+	gate    = sync.NewCond(&sync.Mutex{})
 )
 
 func init() {
 	t := time.Now().UTC()
-	for i := 0; i < feed.Len(); i++ {
-  		feed.Value = change{Time: t}
-  		feed = feed.Next()
-  	}
+	for i := 0; i < journal.Len(); i++ {
+		journal.Value = mutation{Time: t}
+		journal = journal.Next()
+	}
 }
 
-type change struct {
-	Echo      bool      `json:"echo"`
+type mutation struct {
+	Operation string    `json:"operation"`
 	Path      string    `json:"path"`
+	Target    string    `json:"target,omitempty"`
+	Echo      bool      `json:"echo"`
 	StorageID string    `json:"-"`
 	Time      time.Time `json:"-"`
 	UserAgent string    `json:"-"`
@@ -43,73 +43,69 @@ func FileWatch(ctx *App, res http.ResponseWriter, req *http.Request) {
 		SendErrorResult(res, NewError("streaming not supported", 500))
 		return
 	}
-	since, err := time.Parse(time.RFC3339, req.Header.Get("Last-Event-ID"))
+	checkpoint, err := time.Parse(time.RFC3339, req.Header.Get("Last-Event-ID"))
 	if err != nil {
-		since = time.Now()
+		checkpoint = time.Now()
 	}
-	currentDrop := uint64(0)
-	ticker := time.NewTimer(0)
-	defer ticker.Stop()
 	storageID := GenerateID(ctx.Session)
 	userAgent := req.Header.Get("User-Agent")
 	res.Header().Set("Content-Type", "text/event-stream")
 	res.Header().Set("Cache-Control", "no-cache")
 	res.Header().Set("Connection", "keep-alive")
-
-	for {
-		select {
-		case <-req.Context().Done():
+	go func() {
+		<-req.Context().Done()
+		gate.L.Lock()
+		gate.Broadcast()
+		gate.L.Unlock()
+	}()
+	for firstRun := true; ; firstRun = false {
+		gate.L.Lock()
+		if req.Context().Err() != nil {
+			gate.L.Unlock()
 			return
-		case <-ticker.C:
-			flock.Lock()
-			t := feed.Value.(change).Time
-			if nDrop := fdrop.Load(); currentDrop != nDrop {
-				t = time.Now().UTC()
-				fmt.Fprintf(res, "id: %s\nevent: purge\ndata: {}\n\n", t.Format(time.RFC3339Nano))
-				since = t
-				currentDrop = nDrop
-			} else if since.Before(t) {
-				fmt.Fprintf(res, "id: %s\nevent: purge\ndata: {}\n\n", t.Format(time.RFC3339Nano))
-				since = t
-			}
-			feed.Do(func(p any) {
-				el := p.(change)
-				el.Echo = el.UserAgent == userAgent
-				el.Path = strings.TrimPrefix(el.Path, ctx.Session["path"])
-				if el.StorageID != storageID {
-					return
-				} else if since.Equal(el.Time) || since.After(el.Time) {
-					return
-				}
-				data, _ := json.Marshal(el)
-				fmt.Fprintf(res, "id: %s\nevent: dirty\ndata: %s\n\n", el.Time.Format(time.RFC3339Nano), data)
-				since = el.Time
-			})
-			flock.Unlock()
-			flusher.Flush()
-			ticker.Reset(5 * time.Second)
+		} else if firstRun == false {
+			gate.Wait()
 		}
+		changes := make([]mutation, 0, 10)
+		for i, len, curr := 0, journal.Len(), journal.Prev(); i < len; i, curr = i+1, curr.Prev() {
+			el := curr.Value.(mutation)
+			if el.Time.Before(checkpoint) || el.Time.Equal(checkpoint) {
+				break
+			}
+			if el.StorageID != storageID {
+				continue
+			}
+			el.Echo = el.UserAgent == userAgent
+			el.Path = strings.TrimPrefix(el.Path, ctx.Session["path"])
+			el.Target = strings.TrimPrefix(el.Target, ctx.Session["path"])
+			changes = append(changes, el)
+		}
+		gate.L.Unlock()
+		for i := len(changes); i > 0; i-- {
+			el := changes[i-1]
+			data, _ := json.Marshal(el)
+			fmt.Fprintf(res, "id: %s\nevent: mutation\ndata: %s\n\n", el.Time.Format(time.RFC3339Nano), data)
+			checkpoint = changes[0].Time
+		}
+		flusher.Flush()
 	}
 }
 
-func markDirty(ctx *App, req *http.Request, fullpath string) {
-	if !flock.TryLock() {
-		Log.Debug("ls::watch message=cannot_lock_feed", )
-		fdrop.Add(1)
-		return
-	}
-	path, _ := SplitPath(fullpath)
-	cVal := change{
-		Path:      path,
-		StorageID: GenerateID(ctx.Session),
-		Time:      time.Now().UTC(),
-		UserAgent: req.Header.Get("User-Agent"),
-	}
-	pVal := feed.Prev().Value.(change)
-	if pVal.StorageID == cVal.StorageID && pVal.Path == cVal.Path && pVal.UserAgent == cVal.UserAgent {
-		feed = feed.Prev()
-	}
-	feed.Value = cVal
-	feed = feed.Next()
-	flock.Unlock()
+func EventWatch(ctx *App, req *http.Request, op string, path string, target ...string) {
+	t := time.Now()
+	go func() {
+		cVal := mutation{
+			Operation: op,
+			Path:      path,
+			Target:    strings.Join(target, ","),
+			Time:      t.UTC(),
+			StorageID: GenerateID(ctx.Session),
+			UserAgent: req.Header.Get("User-Agent"),
+		}
+		gate.L.Lock()
+		journal.Value = cVal
+		journal = journal.Next()
+		gate.Broadcast()
+		gate.L.Unlock()
+	}()
 }