diff --git a/server/pkg/files/handler_cat.go b/server/pkg/files/handler_cat.go index ca1fb3ef..8e8ccee2 100644 --- a/server/pkg/files/handler_cat.go +++ b/server/pkg/files/handler_cat.go @@ -71,7 +71,7 @@ func FileCat(ctx *App, res http.ResponseWriter, req *http.Request) { if req.Method == http.MethodHead { cmd = "stat" } - op := journal.RecordFile(ctx, req, cmd, path) + op := journal.PublishFS(ctx, req, cmd, path) defer op.Close(res) for _, auth := range Hooks.Get.AuthorisationMiddleware() { diff --git a/server/pkg/files/handler_ls.go b/server/pkg/files/handler_ls.go index 0906a397..81ca9d62 100644 --- a/server/pkg/files/handler_ls.go +++ b/server/pkg/files/handler_ls.go @@ -43,7 +43,7 @@ func FileLs(ctx *App, res http.ResponseWriter, req *http.Request) { SendErrorResult(res, err) return } - op := journal.RecordFile(ctx, req, "ls", path) + op := journal.PublishFS(ctx, req, "ls", path) defer op.Close(res) perms := Metadata{} diff --git a/server/pkg/files/handler_mkdir.go b/server/pkg/files/handler_mkdir.go index 3fb61ba0..84e5c4a2 100644 --- a/server/pkg/files/handler_mkdir.go +++ b/server/pkg/files/handler_mkdir.go @@ -24,7 +24,7 @@ func FileMkdir(ctx *App, res http.ResponseWriter, req *http.Request) { SendErrorResult(res, err) return } - op := journal.RecordFile(ctx, req, "mkdir", path) + op := journal.PublishFS(ctx, req, "mkdir", path) defer op.Close(res) for _, auth := range Hooks.Get.AuthorisationMiddleware() { diff --git a/server/pkg/files/handler_mv.go b/server/pkg/files/handler_mv.go index 195146e7..a5a98e2e 100644 --- a/server/pkg/files/handler_mv.go +++ b/server/pkg/files/handler_mv.go @@ -35,7 +35,7 @@ func FileMv(ctx *App, res http.ResponseWriter, req *http.Request) { SendErrorResult(res, NewError("missing path parameter", 400)) return } - op := journal.RecordFile(ctx, req, "mv", from, to) + op := journal.PublishFS(ctx, req, "mv", from, to) defer op.Close(res) for _, auth := range Hooks.Get.AuthorisationMiddleware() { diff --git a/server/pkg/files/handler_rm.go b/server/pkg/files/handler_rm.go index a4ab0435..392f9739 100644 --- a/server/pkg/files/handler_rm.go +++ b/server/pkg/files/handler_rm.go @@ -24,7 +24,7 @@ func FileRm(ctx *App, res http.ResponseWriter, req *http.Request) { SendErrorResult(res, err) return } - op := journal.RecordFile(ctx, req, "rm", path) + op := journal.PublishFS(ctx, req, "rm", path) defer op.Close(res) for _, auth := range Hooks.Get.AuthorisationMiddleware() { diff --git a/server/pkg/files/handler_save.go b/server/pkg/files/handler_save.go index 54aa0915..55001e10 100644 --- a/server/pkg/files/handler_save.go +++ b/server/pkg/files/handler_save.go @@ -40,7 +40,7 @@ func FileSave(ctx *App, res http.ResponseWriter, req *http.Request) { SendErrorResult(res, err) return } - op := journal.RecordFile(ctx, req, "save", path) + op := journal.PublishFS(ctx, req, "save", path) defer op.Close(res) if permissions.CanEdit(ctx) == false { diff --git a/server/pkg/files/handler_touch.go b/server/pkg/files/handler_touch.go index 948b38f1..5f7957bd 100644 --- a/server/pkg/files/handler_touch.go +++ b/server/pkg/files/handler_touch.go @@ -24,7 +24,7 @@ func FileTouch(ctx *App, res http.ResponseWriter, req *http.Request) { SendErrorResult(res, err) return } - op := journal.RecordFile(ctx, req, "touch", path) + op := journal.PublishFS(ctx, req, "touch", path) defer op.Close(res) for _, auth := range Hooks.Get.AuthorisationMiddleware() { diff --git a/server/pkg/files/handler_watch.go b/server/pkg/files/handler_watch.go index 050982ec..feef26af 100644 --- a/server/pkg/files/handler_watch.go +++ b/server/pkg/files/handler_watch.go @@ -8,9 +8,10 @@ import ( "time" . "github.com/mickael-kerjean/filestash/server/pkg/core" - . "github.com/mickael-kerjean/filestash/server/pkg/journal" . "github.com/mickael-kerjean/filestash/server/pkg/kernel" . "github.com/mickael-kerjean/filestash/server/pkg/utils" + + "github.com/mickael-kerjean/filestash/server/pkg/journal" ) const heartbeatPeriod = 15 * time.Second @@ -42,15 +43,19 @@ func FileWatch(ctx *App, res http.ResponseWriter, req *http.Request) { res.Header().Set("Connection", "keep-alive") flusher.Flush() - events := Listen[FileOp](req.Context(), checkpoint, func(el Observation[FileOp]) bool { - return el.Payload.StorageID == storageID && el.Payload.Mutation && el.Done && el.Error == nil - }) + stream := journal.Subscribe[journal.FileOp]( + req.Context(), + checkpoint, + func(el journal.Observation[journal.FileOp]) bool { + return el.Payload.StorageID == storageID && el.Payload.Mutation && el.Done && el.Error == nil + }, + ) heartbeat := time.NewTicker(heartbeatPeriod) defer heartbeat.Stop() for { select { - case changes, more := <-events: + case changes, more := <-stream: if !more { return } diff --git a/server/pkg/journal/index.go b/server/pkg/journal/index.go index 4a204e1c..854ae39a 100644 --- a/server/pkg/journal/index.go +++ b/server/pkg/journal/index.go @@ -11,7 +11,7 @@ var ( ) type Payload interface { - FileOp | SessionOp | Nop + FileOp | SessionOp | EventOp | Nop } type FileOp struct { @@ -26,4 +26,6 @@ type SessionOp struct { Operation string `json:"operation"` } +type EventOp map[string]any + type Nop struct{} diff --git a/server/pkg/journal/publish.go b/server/pkg/journal/publish.go index 6afc5a14..14fcc657 100644 --- a/server/pkg/journal/publish.go +++ b/server/pkg/journal/publish.go @@ -12,7 +12,7 @@ import ( "github.com/mickael-kerjean/filestash/server/pkg/tracer" ) -func RecordFile(ctx *App, req *http.Request, topic string, path string, target ...string) Observation[FileOp] { +func PublishFS(ctx *App, req *http.Request, topic string, path string, target ...string) Observation[FileOp] { return NewObservation(Observation[FileOp]{ At: time.Now().UTC(), Kind: "fs", @@ -28,7 +28,7 @@ func RecordFile(ctx *App, req *http.Request, topic string, path string, target . }) } -func RecordSession(ctx *App, req *http.Request, cmd string) Observation[SessionOp] { +func PublishSession(ctx *App, req *http.Request, cmd string) Observation[SessionOp] { return NewObservation(Observation[SessionOp]{ At: time.Now().UTC(), Kind: "session", @@ -40,6 +40,16 @@ func RecordSession(ctx *App, req *http.Request, cmd string) Observation[SessionO }) } +func PublishEvent(ctx *App, req *http.Request, data EventOp) Observation[EventOp] { + return NewObservation(Observation[EventOp]{ + At: time.Now().UTC(), + Kind: "event", + Payload: data, + Trace: tracer.Extract(req), + Emit: emit[EventOp], + }) +} + func emit[T Payload](val Observation[T]) { gate.L.Lock() journal.Set(val) diff --git a/server/pkg/journal/subscribe.go b/server/pkg/journal/subscribe.go index de8a6323..4340f477 100644 --- a/server/pkg/journal/subscribe.go +++ b/server/pkg/journal/subscribe.go @@ -6,7 +6,7 @@ import ( "time" ) -func Listen[T Payload](ctx context.Context, since time.Time, filter func(Observation[T]) bool) <-chan []Observation[T] { +func Subscribe[T Payload](ctx context.Context, since time.Time, filter func(Observation[T]) bool) <-chan []Observation[T] { out := make(chan []Observation[T]) go func() { for firstRun, checkpoint := true, since; ; firstRun = false { diff --git a/server/pkg/session/handler.go b/server/pkg/session/handler.go index f1641ab8..defea045 100644 --- a/server/pkg/session/handler.go +++ b/server/pkg/session/handler.go @@ -33,7 +33,7 @@ type Session struct { } func SessionGet(ctx *App, res http.ResponseWriter, req *http.Request) { - op := journal.RecordSession(ctx, req, "get") + op := journal.PublishSession(ctx, req, "get") defer op.Close(res) r := Session{ IsAuth: false, @@ -59,7 +59,7 @@ func SessionGet(ctx *App, res http.ResponseWriter, req *http.Request) { } func SessionAuthenticate(ctx *App, res http.ResponseWriter, req *http.Request) { - op := journal.RecordSession(ctx, req, "auth") + op := journal.PublishSession(ctx, req, "auth") defer op.Close(res) ctx.Body["timestamp"] = time.Now().Format(time.RFC3339) session := MapStringInterfaceToMapStringString(ctx.Body) @@ -124,7 +124,7 @@ func SessionAuthenticate(ctx *App, res http.ResponseWriter, req *http.Request) { } func SessionLogout(ctx *App, res http.ResponseWriter, req *http.Request) { - op := journal.RecordSession(ctx, req, "logout") + op := journal.PublishSession(ctx, req, "logout") defer op.Close(res) go func() { // user typically expect the logout to feel instant but in our case we still need to make sure @@ -159,7 +159,7 @@ func SessionLogout(ctx *App, res http.ResponseWriter, req *http.Request) { } func SessionOAuthBackend(ctx *App, res http.ResponseWriter, req *http.Request) { - op := journal.RecordSession(ctx, req, "oauth") + op := journal.PublishSession(ctx, req, "oauth") defer op.Close(res) vars := mux.Vars(req) a := map[string]string{ @@ -198,7 +198,7 @@ func SessionOAuthBackend(ctx *App, res http.ResponseWriter, req *http.Request) { } func SessionAuthMiddleware(ctx *App, res http.ResponseWriter, req *http.Request) { - op := journal.RecordSession(ctx, req, "middleware") + op := journal.PublishSession(ctx, req, "middleware") defer op.Close(res) SSOCookieName := "ssoref"