feat(KLEF-160): Notification system

internal/bootstrap/container.go

@@ -23,10 +23,14 @@ import (
 	deploymentshttp "github.com/kleffio/platform/internal/core/deployments/adapters/http"
 	deploymentspersistence "github.com/kleffio/platform/internal/core/deployments/adapters/persistence"
 	deploymentscommands "github.com/kleffio/platform/internal/core/deployments/application/commands"
+	notificationshttp "github.com/kleffio/platform/internal/core/notifications/adapters/http"
+	notificationspersistence "github.com/kleffio/platform/internal/core/notifications/adapters/persistence"
+	notificationsapp "github.com/kleffio/platform/internal/core/notifications/application"
 	nodeshttp "github.com/kleffio/platform/internal/core/nodes/adapters/http"
 	nodespersistence "github.com/kleffio/platform/internal/core/nodes/adapters/persistence"
 	nodesapp "github.com/kleffio/platform/internal/core/nodes/application"
 	organizationshttp "github.com/kleffio/platform/internal/core/organizations/adapters/http"
+	organizationspersistence "github.com/kleffio/platform/internal/core/organizations/adapters/persistence"
 	pluginhttp "github.com/kleffio/platform/internal/core/plugins/adapters/http"
 	pluginpersistence "github.com/kleffio/platform/internal/core/plugins/adapters/persistence"
 	pluginregistry "github.com/kleffio/platform/internal/core/plugins/adapters/registry"
@@ -58,19 +62,22 @@ type Container struct {
 	PluginManager *pluginapplication.Manager
 
 	// HTTP handler groups per domain module
-	AuthHandler          *pluginhttp.AuthHandler
-	SetupHandler         *pluginhttp.SetupHandler
-	CatalogHandler       *cataloghttp.Handler
-	OrganizationsHandler *organizationshttp.Handler
-	ProjectsHandler      *projectshttp.Handler
-	WorkloadsHandler     *workloadshttp.Handler
-	DeploymentsHandler   *deploymentshttp.Handler
-	NodesHandler         *nodeshttp.Handler
-	BillingHandler       *billinghttp.Handler
-	UsageHandler         *usagehttp.Handler
-	AuditHandler         *audithttp.Handler
-	AdminHandler         *adminhttp.Handler
-	PluginsHandler       *pluginhttp.Handler
+	AuthHandler           *pluginhttp.AuthHandler
+	SetupHandler          *pluginhttp.SetupHandler
+	CatalogHandler        *cataloghttp.Handler
+	OrganizationsHandler  *organizationshttp.Handler
+	ProjectsHandler       *projectshttp.Handler
+	WorkloadsHandler      *workloadshttp.Handler
+	DeploymentsHandler    *deploymentshttp.Handler
+	NodesHandler          *nodeshttp.Handler
+	BillingHandler        *billinghttp.Handler
+	UsageHandler          *usagehttp.Handler
+	AuditHandler          *audithttp.Handler
+	AdminHandler          *adminhttp.Handler
+	PluginsHandler        *pluginhttp.Handler
+	NotificationsHandler  *notificationshttp.Handler
+	NotificationService   *notificationsapp.Service
+	NotificationHub       *notificationsapp.Hub
 }
 
 // NewContainer wires all dependencies and returns the composition root.
@@ -129,6 +136,7 @@ func NewContainer(cfg *Config, logger *slog.Logger) (*Container, error) {
 	nodeStore := nodespersistence.NewPostgresNodeStore(db)
 	nodeVerifier := nodesapp.NewTokenVerifier(nodeStore)
 
+	orgStore := organizationspersistence.NewPostgresOrgStore(db)
 	projectsStore := projectspersistence.NewPostgresProjectStore(db)
 	workloadsStore := workloadspersistence.NewPostgresStore(db)
 
@@ -141,6 +149,11 @@ func NewContainer(cfg *Config, logger *slog.Logger) (*Container, error) {
 	}
 
 	bus := events.New()
+
+	notificationStore := notificationspersistence.NewPostgresNotificationStore(db)
+	notificationHub := notificationsapp.NewHub()
+	notificationSvc := notificationsapp.NewService(notificationStore, notificationHub, logger)
+
 	provisionHandler := workloadcmd.NewProvisionWorkloadHandler(workloadsStore, projectsStore, queuePublisher, catalogStore, logger)
 	workloadAction := workloadcmd.NewWorkloadActionHandler(workloadsStore, projectsStore, queuePublisher, logger)
 
@@ -156,16 +169,19 @@ func NewContainer(cfg *Config, logger *slog.Logger) (*Container, error) {
 		AuthHandler:          pluginhttp.NewAuthHandler(pluginMgr, logger),
 		SetupHandler:         pluginhttp.NewSetupHandler(pluginMgr, catalogRegistry, logger),
 		CatalogHandler:       cataloghttp.NewHandler(catalogStore, logger),
-		OrganizationsHandler: organizationshttp.NewHandler(logger),
+		OrganizationsHandler: organizationshttp.NewHandler(orgStore, notificationSvc, logger),
 		DeploymentsHandler:   deploymentshttp.NewHandler(createDeployment, serverAction, deploymentStore, cfg.SecretKey, logger),
-		ProjectsHandler:      projectshttp.NewHandler(projectsStore, logger),
-		WorkloadsHandler:     workloadshttp.NewHandler(projectsStore, workloadsStore, provisionHandler, workloadAction, bus, logger),
+		ProjectsHandler:      projectshttp.NewHandler(projectsStore, orgStore, logger),
+		WorkloadsHandler:     workloadshttp.NewHandler(projectsStore, orgStore, workloadsStore, provisionHandler, workloadAction, bus, logger),
 		NodesHandler:         nodeshttp.NewHandler(nodeStore, logger),
 		BillingHandler:       billinghttp.NewHandler(logger),
 		UsageHandler:         usagehttp.NewHandler(logger),
 		AuditHandler:         audithttp.NewHandler(logger),
 		AdminHandler:         adminhttp.NewHandler(logger),
 		PluginsHandler:       pluginhttp.NewHandler(pluginMgr, catalogRegistry, logger),
+		NotificationsHandler: notificationshttp.NewHandler(notificationSvc, notificationHub, logger),
+		NotificationService:  notificationSvc,
+		NotificationHub:      notificationHub,
 	}, nil
 }
 

internal/bootstrap/http.go

@@ -69,6 +69,7 @@ func buildRouter(c *Container) http.Handler {
 		c.BillingHandler.RegisterRoutes(r)
 		c.UsageHandler.RegisterRoutes(r)
 		c.AuditHandler.RegisterRoutes(r)
+		c.NotificationsHandler.RegisterRoutes(r)
 
 		// Admin routes — additionally require the "admin" role.
 		r.Group(func(r chi.Router) {

internal/core/notifications/adapters/http/handler.go

@@ -0,0 +1,213 @@
+// Package http exposes REST endpoints and an SSE stream for the notifications module.
+package http
+
+import (
+	"encoding/json"
+	"fmt"
+	"log/slog"
+	"net/http"
+	"strconv"
+	"time"
+
+	"github.com/go-chi/chi/v5"
+	"github.com/kleffio/platform/internal/core/notifications/application"
+	"github.com/kleffio/platform/internal/core/notifications/domain"
+	"github.com/kleffio/platform/internal/shared/middleware"
+)
+
+const basePath = "/api/v1/notifications"
+
+// Handler groups all HTTP endpoints for the notifications module.
+type Handler struct {
+	svc    *application.Service
+	hub    *application.Hub
+	logger *slog.Logger
+}
+
+// NewHandler creates a Handler.
+func NewHandler(svc *application.Service, hub *application.Hub, logger *slog.Logger) *Handler {
+	return &Handler{svc: svc, hub: hub, logger: logger}
+}
+
+// RegisterRoutes attaches all notification routes to the provided router.
+// All routes require a valid JWT (caller must already be wrapped in RequireAuth).
+func (h *Handler) RegisterRoutes(r chi.Router) {
+	r.Get(basePath, h.list)
+	r.Get(basePath+"/unread-count", h.unreadCount)
+	r.Get(basePath+"/stream", h.stream)
+	r.Post(basePath+"/read-all", h.markAllRead)
+	r.Patch(basePath+"/{id}/read", h.markRead)
+	r.Delete(basePath+"/{id}", h.delete)
+}
+
+// ── List ──────────────────────────────────────────────────────────────────────
+
+func (h *Handler) list(w http.ResponseWriter, r *http.Request) {
+	claims, ok := middleware.ClaimsFromContext(r.Context())
+	if !ok {
+		writeJSON(w, http.StatusUnauthorized, errBody("unauthorized"))
+		return
+	}
+
+	q := r.URL.Query()
+	unreadOnly := q.Get("unread") == "true"
+	limit, _ := strconv.Atoi(q.Get("limit"))
+	offset, _ := strconv.Atoi(q.Get("offset"))
+
+	notifications, err := h.svc.List(r.Context(), claims.Subject, domain.ListFilter{
+		UnreadOnly: unreadOnly,
+		Limit:      limit,
+		Offset:     offset,
+	})
+	if err != nil {
+		h.logger.Error("list notifications", "error", err)
+		writeJSON(w, http.StatusInternalServerError, errBody("failed to list notifications"))
+		return
+	}
+	if notifications == nil {
+		notifications = []*domain.Notification{}
+	}
+	writeJSON(w, http.StatusOK, map[string]any{"notifications": notifications})
+}
+
+// ── Unread count ──────────────────────────────────────────────────────────────
+
+func (h *Handler) unreadCount(w http.ResponseWriter, r *http.Request) {
+	claims, ok := middleware.ClaimsFromContext(r.Context())
+	if !ok {
+		writeJSON(w, http.StatusUnauthorized, errBody("unauthorized"))
+		return
+	}
+
+	count, err := h.svc.CountUnread(r.Context(), claims.Subject)
+	if err != nil {
+		h.logger.Error("count unread notifications", "error", err)
+		writeJSON(w, http.StatusInternalServerError, errBody("failed to count unread notifications"))
+		return
+	}
+	writeJSON(w, http.StatusOK, map[string]int{"unread_count": count})
+}
+
+// ── Mark single read ──────────────────────────────────────────────────────────
+
+func (h *Handler) markRead(w http.ResponseWriter, r *http.Request) {
+	claims, ok := middleware.ClaimsFromContext(r.Context())
+	if !ok {
+		writeJSON(w, http.StatusUnauthorized, errBody("unauthorized"))
+		return
+	}
+
+	id := chi.URLParam(r, "id")
+	if err := h.svc.MarkRead(r.Context(), id, claims.Subject); err != nil {
+		h.logger.Error("mark notification read", "error", err)
+		writeJSON(w, http.StatusInternalServerError, errBody("failed to mark notification as read"))
+		return
+	}
+	w.WriteHeader(http.StatusNoContent)
+}
+
+// ── Mark all read ─────────────────────────────────────────────────────────────
+
+func (h *Handler) markAllRead(w http.ResponseWriter, r *http.Request) {
+	claims, ok := middleware.ClaimsFromContext(r.Context())
+	if !ok {
+		writeJSON(w, http.StatusUnauthorized, errBody("unauthorized"))
+		return
+	}
+
+	if err := h.svc.MarkAllRead(r.Context(), claims.Subject); err != nil {
+		h.logger.Error("mark all notifications read", "error", err)
+		writeJSON(w, http.StatusInternalServerError, errBody("failed to mark all notifications as read"))
+		return
+	}
+	w.WriteHeader(http.StatusNoContent)
+}
+
+// ── Delete ────────────────────────────────────────────────────────────────────
+
+func (h *Handler) delete(w http.ResponseWriter, r *http.Request) {
+	claims, ok := middleware.ClaimsFromContext(r.Context())
+	if !ok {
+		writeJSON(w, http.StatusUnauthorized, errBody("unauthorized"))
+		return
+	}
+
+	id := chi.URLParam(r, "id")
+	if err := h.svc.Delete(r.Context(), id, claims.Subject); err != nil {
+		h.logger.Error("delete notification", "error", err)
+		writeJSON(w, http.StatusInternalServerError, errBody("failed to delete notification"))
+		return
+	}
+	w.WriteHeader(http.StatusNoContent)
+}
+
+// ── SSE stream ────────────────────────────────────────────────────────────────
+
+// stream opens a Server-Sent Events connection for the authenticated user.
+// On connect it sends the current unread count, then streams new notifications
+// as they arrive. A heartbeat comment is sent every 30 s to keep the connection
+// alive through proxies.
+func (h *Handler) stream(w http.ResponseWriter, r *http.Request) {
+	claims, ok := middleware.ClaimsFromContext(r.Context())
+	if !ok {
+		http.Error(w, "unauthorized", http.StatusUnauthorized)
+		return
+	}
+
+	flusher, ok := w.(http.Flusher)
+	if !ok {
+		http.Error(w, "streaming not supported", http.StatusInternalServerError)
+		return
+	}
+
+	w.Header().Set("Content-Type", "text/event-stream")
+	w.Header().Set("Cache-Control", "no-cache")
+	w.Header().Set("Connection", "keep-alive")
+	w.Header().Set("X-Accel-Buffering", "no") // disable nginx buffering
+
+	// Send initial connected event with the current unread count.
+	count, _ := h.svc.CountUnread(r.Context(), claims.Subject)
+	fmt.Fprintf(w, "event: connected\ndata: {\"unread_count\":%d}\n\n", count)
+	flusher.Flush()
+
+	ch := h.hub.Subscribe(claims.Subject)
+	defer h.hub.Unsubscribe(claims.Subject, ch)
+
+	ticker := time.NewTicker(30 * time.Second)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-r.Context().Done():
+			return
+
+		case <-ticker.C:
+			// Heartbeat — keeps the connection alive through idle-timeout proxies.
+			fmt.Fprintf(w, ": heartbeat\n\n")
+			flusher.Flush()
+
+		case n, open := <-ch:
+			if !open {
+				return
+			}
+			data, err := json.Marshal(n)
+			if err != nil {
+				continue
+			}
+			fmt.Fprintf(w, "event: notification\ndata: %s\n\n", data)
+			flusher.Flush()
+		}
+	}
+}
+
+// ── Helpers ───────────────────────────────────────────────────────────────────
+
+func errBody(msg string) map[string]string {
+	return map[string]string{"error": msg}
+}
+
+func writeJSON(w http.ResponseWriter, status int, body any) {
+	w.Header().Set("Content-Type", "application/json")
+	w.WriteHeader(status)
+	_ = json.NewEncoder(w).Encode(body)
+}