feat: WebSocket based implementation of server sent events for cache busting (#527)

* rough implementation of WS based event system for server side notifications of mutation

* fix test construction

* fix deadlock on event bus

* disable linter error

* add item mutation events

* remove old event bus code

* refactor event system to use composables

* refresh items table when new item is added

* fix create form errors

* cleanup unnecessary calls

* fix importer erorrs + limit fn calls on import

backend/app/api/app.go

@@ -4,6 +4,7 @@ import (
 	"time"
 
 	"github.com/hay-kot/homebox/backend/internal/core/services"
+	"github.com/hay-kot/homebox/backend/internal/core/services/reporting/eventbus"
 	"github.com/hay-kot/homebox/backend/internal/data/ent"
 	"github.com/hay-kot/homebox/backend/internal/data/repo"
 	"github.com/hay-kot/homebox/backend/internal/sys/config"
@@ -18,6 +19,7 @@ type app struct {
 	server   *server.Server
 	repos    *repo.AllRepos
 	services *services.AllServices
+	bus      *eventbus.EventBus
 }
 
 func new(conf *config.Config) *app {

backend/app/api/handlers/v1/controller.go

@@ -1,12 +1,18 @@
 package v1
 
 import (
+	"fmt"
 	"net/http"
 
+	"github.com/google/uuid"
 	"github.com/hay-kot/homebox/backend/internal/core/services"
+	"github.com/hay-kot/homebox/backend/internal/core/services/reporting/eventbus"
 	"github.com/hay-kot/homebox/backend/internal/data/repo"
 	"github.com/hay-kot/httpkit/errchain"
 	"github.com/hay-kot/httpkit/server"
+	"github.com/rs/zerolog/log"
+
+	"github.com/olahol/melody"
 )
 
 type Results[T any] struct {
@@ -49,6 +55,7 @@ type V1Controller struct {
 	maxUploadSize     int64
 	isDemo            bool
 	allowRegistration bool
+	bus               *eventbus.EventBus
 }
 
 type (
@@ -77,11 +84,12 @@ func BaseUrlFunc(prefix string) func(s string) string {
 	}
 }
 
-func NewControllerV1(svc *services.AllServices, repos *repo.AllRepos, options ...func(*V1Controller)) *V1Controller {
+func NewControllerV1(svc *services.AllServices, repos *repo.AllRepos, bus *eventbus.EventBus, options ...func(*V1Controller)) *V1Controller {
 	ctrl := &V1Controller{
 		repo:              repos,
 		svc:               svc,
 		allowRegistration: true,
+		bus:               bus,
 	}
 
 	for _, opt := range options {
@@ -110,3 +118,42 @@ func (ctrl *V1Controller) HandleBase(ready ReadyFunc, build Build) errchain.Hand
 		})
 	}
 }
+
+func (ctrl *V1Controller) HandleCacheWS() errchain.HandlerFunc {
+	m := melody.New()
+
+	m.HandleConnect(func(s *melody.Session) {
+		auth := services.NewContext(s.Request.Context())
+		s.Set("gid", auth.GID)
+	})
+
+	factory := func(e string) func(data any) {
+		return func(data any) {
+			eventData, ok := data.(eventbus.GroupMutationEvent)
+			if !ok {
+				log.Log().Msgf("invalid event data: %v", data)
+				return
+			}
+
+			jsonStr := fmt.Sprintf(`{"event": "%s"}`, e)
+
+			_ = m.BroadcastFilter([]byte(jsonStr), func(s *melody.Session) bool {
+				groupIDStr, ok := s.Get("gid")
+				if !ok {
+					return false
+				}
+
+				GID := groupIDStr.(uuid.UUID)
+				return GID == eventData.GID
+			})
+		}
+	}
+
+	ctrl.bus.Subscribe(eventbus.EventLabelMutation, factory("label.mutation"))
+	ctrl.bus.Subscribe(eventbus.EventLocationMutation, factory("location.mutation"))
+	ctrl.bus.Subscribe(eventbus.EventItemMutation, factory("item.mutation"))
+
+	return func(w http.ResponseWriter, r *http.Request) error {
+		return m.HandleRequest(w, r)
+	}
+}

backend/app/api/main.go

@@ -14,6 +14,7 @@ import (
 	"github.com/go-chi/chi/v5/middleware"
 
 	"github.com/hay-kot/homebox/backend/internal/core/services"
+	"github.com/hay-kot/homebox/backend/internal/core/services/reporting/eventbus"
 	"github.com/hay-kot/homebox/backend/internal/data/ent"
 	"github.com/hay-kot/homebox/backend/internal/data/migrations"
 	"github.com/hay-kot/homebox/backend/internal/data/repo"
@@ -116,8 +117,9 @@ func run(cfg *config.Config) error {
 		return err
 	}
 
+	app.bus = eventbus.New()
 	app.db = c
-	app.repos = repo.New(c, cfg.Storage.Data)
+	app.repos = repo.New(c, app.bus, cfg.Storage.Data)
 	app.services = services.New(
 		app.repos,
 		services.WithAutoIncrementAssetID(cfg.Options.AutoIncrementAssetID),
@@ -150,6 +152,8 @@ func run(cfg *config.Config) error {
 	// =========================================================================
 	// Start Reoccurring Tasks
 
+	go app.bus.Run()
+
 	go app.startBgTask(time.Duration(24)*time.Hour, func() {
 		_, err := app.repos.AuthTokens.PurgeExpiredTokens(context.Background())
 		if err != nil {

backend/app/api/routes.go

@@ -51,6 +51,7 @@ func (a *app) mountRoutes(r *chi.Mux, chain *errchain.ErrChain, repos *repo.AllR
 	v1Ctrl := v1.NewControllerV1(
 		a.services,
 		a.repos,
+    a.bus,
 		v1.WithMaxUploadSize(a.conf.Web.MaxUploadSize),
 		v1.WithRegistration(a.conf.Options.AllowRegistration),
 		v1.WithDemoStatus(a.conf.Demo), // Disable Password Change in Demo Mode
@@ -70,6 +71,7 @@ func (a *app) mountRoutes(r *chi.Mux, chain *errchain.ErrChain, repos *repo.AllR
 		a.mwRoles(RoleModeOr, authroles.RoleUser.String()),
 	}
 
+	r.Get(v1Base("/ws/events"), chain.ToHandlerFunc(v1Ctrl.HandleCacheWS(), userMW...))
 	r.Get(v1Base("/users/self"), chain.ToHandlerFunc(v1Ctrl.HandleUserSelf(), userMW...))
 	r.Put(v1Base("/users/self"), chain.ToHandlerFunc(v1Ctrl.HandleUserSelfUpdate(), userMW...))
 	r.Delete(v1Base("/users/self"), chain.ToHandlerFunc(v1Ctrl.HandleUserSelfDelete(), userMW...))
@@ -153,7 +155,6 @@ func (a *app) mountRoutes(r *chi.Mux, chain *errchain.ErrChain, repos *repo.AllR
 	r.Get(v1Base("/reporting/bill-of-materials"), chain.ToHandlerFunc(v1Ctrl.HandleBillOfMaterialsExport(), userMW...))
 
 	r.NotFound(chain.ToHandlerFunc(notFoundHandler()))
-
 }
 
 func registerMimes() {