github/filestash
4
0
Fork 0
mirror of https://github.com/mickael-kerjean/filestash.git synced 2026-10-19 16:14:39 +00:00
Code Issues Releases Wiki Activity

chore (refactoring): improve journal public api

Browse Source
This commit is contained in:
MickaelK
2026-09-17 21:23:03 +10:00
parent 5a44d82e5d
commit 9662ee565a
12 changed files with 38 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/pkg/files/handler_cat.go
+1 -1
View File
@@ -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() {
server/pkg/files/handler_ls.go
+1 -1
View File
@@ -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{}
server/pkg/files/handler_mkdir.go
+1 -1
View File
@@ -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() {
server/pkg/files/handler_mv.go
+1 -1
View File
@@ -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() {
server/pkg/files/handler_rm.go
+1 -1
View File
@@ -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() {
server/pkg/files/handler_save.go
+1 -1
View File
@@ -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 {
server/pkg/files/handler_touch.go
+1 -1
View File
@@ -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() {
server/pkg/files/handler_watch.go
+10 -5
View File
@@ -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
}
server/pkg/journal/index.go
+3 -1
View File
@@ -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{}
server/pkg/journal/publish.go
+12 -2
View File
@@ -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)
server/pkg/journal/subscribe.go
+1 -1
View File
@@ -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 {
server/pkg/session/handler.go
+5 -5
View File
@@ -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"