feat: expose security notification APIs

Wires notification and two-factor services into the backend and consumer, adds the authenticated notification feed/preferences endpoints, and exposes the public 2FA login verification endpoint.

cmd/backend/main.go

@@ -42,6 +42,7 @@ import (
 	idempotencyapp "github.com/warmbly/warmbly/internal/app/idempotency"
 	"github.com/warmbly/warmbly/internal/app/integration"
 	"github.com/warmbly/warmbly/internal/app/leadsync"
+	"github.com/warmbly/warmbly/internal/app/notification"
 	"github.com/warmbly/warmbly/internal/app/organization"
 	"github.com/warmbly/warmbly/internal/app/passkey"
 	"github.com/warmbly/warmbly/internal/app/placement"
@@ -57,6 +58,7 @@ import (
 	"github.com/warmbly/warmbly/internal/app/template"
 	"github.com/warmbly/warmbly/internal/app/token"
 	"github.com/warmbly/warmbly/internal/app/trial"
+	"github.com/warmbly/warmbly/internal/app/twofa"
 	"github.com/warmbly/warmbly/internal/app/tz"
 	"github.com/warmbly/warmbly/internal/app/unibox"
 	"github.com/warmbly/warmbly/internal/app/user"
@@ -190,6 +192,8 @@ func main() {
 	var warmupRoutingRepoForHandler repository.WarmupRoutingRepository
 	var webhookServiceForHandler webhook.Service
 	var integrationServiceForHandler integration.Service
+	var notificationService notification.Service
+	var twofaService twofa.Service
 	var contactRepoForHandler repository.ContactRepository
 	var attachmentRepoForHandler repository.AttachmentRepository
 	var leadSyncServiceForHandler leadsync.Service
@@ -587,6 +591,11 @@ func main() {
 			userRepostory,
 			userService,
 		)
+		// TOTP 2FA: the secret is sealed with a server-wide key (the per-user DEK
+		// is unreachable at login time). Wire the challenger into auth so the login
+		// gate can issue a pending challenge.
+		twofaService = twofa.NewService(repository.NewTOTPRepository(primaryDB.Pool), userRepostory, tokenService, cache, twofa.DeriveKey(authCfg.TwoFASecret))
+		authService.WireTwoFA(twofaService)
 		var passkeyErr error
 		passkeyService, passkeyErr = passkey.New(passkey.Deps{
 			Repo:          webauthnRepository,
@@ -855,6 +864,11 @@ func main() {
 			orgs:     organizationRepository,
 		})
 		integrationServiceForHandler.SetPublisher(streamingPublisher)
+		// In-app notifications: API reads/writes happen here; also wire the gate
+		// onto the backend's advanced service (deliverability webhooks can ingest
+		// here too).
+		notificationService = notification.NewService(repository.NewNotificationRepository(primaryDB.Pool), streamingPublisher)
+		advancedService.WireNotifier(notificationService)
 		emailSender := tasks.NewEmailSender(emailRepostory, eventsPublisher)
 		tasksService = tasks.NewService(
 			tasksClient,
@@ -985,9 +999,11 @@ func main() {
 		TagService:      tagService,
 		CategoryService: categoryService,
 
-		TzService:     tzService,
-		SocketService: socketService,
-		TasksService:  tasksService,
+		TzService:           tzService,
+		SocketService:       socketService,
+		TasksService:        tasksService,
+		NotificationService: notificationService,
+		TwoFAService:        twofaService,
 
 		// API Keys
 		APIKeyService: apiKeyService,

cmd/consumer/main.go

@@ -14,6 +14,7 @@ import (
 	"github.com/warmbly/warmbly/internal/app/cipher"
 	jobs "github.com/warmbly/warmbly/internal/app/consumer"
 	"github.com/warmbly/warmbly/internal/app/integration"
+	"github.com/warmbly/warmbly/internal/app/notification"
 	warmupapp "github.com/warmbly/warmbly/internal/app/warmup"
 	"github.com/warmbly/warmbly/internal/app/webhook"
 	workerapp "github.com/warmbly/warmbly/internal/app/worker"
@@ -222,6 +223,12 @@ func main() {
 		warmupService,
 	)
 	advancedService.WireDispatcher(webhookService)
+	// In-app notifications: the reply/bounce/complaint gate fires in THIS
+	// process (inbox ingest + deliverability ingest run in the consumer), so the
+	// notifier must be wired here. Missing this = notifications silently never
+	// created.
+	notificationService := notification.NewService(repository.NewNotificationRepository(primaryDB.Pool), streamingPublisher)
+	advancedService.WireNotifier(notificationService)
 
 	// Events publisher — wraps the existing Kafka producer in an EventBus,
 	// wraps Avrov2 in a Codec. Once EVENTBUS_PROVIDER=nats is exercised in

internal/api/handler/handler.go

@@ -21,6 +21,7 @@ import (
 	"github.com/warmbly/warmbly/internal/app/group"
 	"github.com/warmbly/warmbly/internal/app/integration"
 	"github.com/warmbly/warmbly/internal/app/leadsync"
+	"github.com/warmbly/warmbly/internal/app/notification"
 	"github.com/warmbly/warmbly/internal/app/organization"
 	"github.com/warmbly/warmbly/internal/app/passkey"
 	"github.com/warmbly/warmbly/internal/app/placement"
@@ -34,6 +35,7 @@ import (
 	"github.com/warmbly/warmbly/internal/app/template"
 	"github.com/warmbly/warmbly/internal/app/token"
 	"github.com/warmbly/warmbly/internal/app/trial"
+	"github.com/warmbly/warmbly/internal/app/twofa"
 	"github.com/warmbly/warmbly/internal/app/tz"
 	"github.com/warmbly/warmbly/internal/app/unibox"
 	"github.com/warmbly/warmbly/internal/app/user"
@@ -67,9 +69,11 @@ type Handler struct {
 	TagService      group.GroupService
 	CategoryService group.GroupService
 
-	TzService     tz.TzService
-	SocketService socket.SocketService
-	TasksService  tasks.TasksService
+	TzService           tz.TzService
+	SocketService       socket.SocketService
+	TasksService        tasks.TasksService
+	NotificationService notification.Service
+	TwoFAService        twofa.Service
 
 	// New services
 	APIKeyService    apikey.APIKeyService

internal/api/handler/notification.go

@@ -0,0 +1,109 @@
+package handler
+
+import (
+	"net/http"
+	"strconv"
+
+	"github.com/gin-gonic/gin"
+	"github.com/google/uuid"
+
+	"github.com/warmbly/warmbly/internal/api/middleware"
+	"github.com/warmbly/warmbly/internal/errx"
+	"github.com/warmbly/warmbly/internal/models"
+)
+
+func notifActor(c *gin.Context) (uuid.UUID, bool) {
+	id, err := uuid.Parse(middleware.GetUserID(c))
+	if err != nil {
+		errx.JSON(c, errx.New(errx.BadRequest, "invalid user"))
+		return uuid.Nil, false
+	}
+	return id, true
+}
+
+// GetNotificationPreferences returns the caller's notification preferences
+// (merged over defaults).
+func (h *Handler) GetNotificationPreferences(c *gin.Context) {
+	uid, ok := notifActor(c)
+	if !ok {
+		return
+	}
+	prefs, xerr := h.NotificationService.GetPreferences(c.Request.Context(), uid)
+	if xerr != nil {
+		errx.JSON(c, xerr)
+		return
+	}
+	c.JSON(http.StatusOK, gin.H{"preferences": prefs})
+}
+
+// UpdateNotificationPreferences replaces the caller's notification preferences.
+func (h *Handler) UpdateNotificationPreferences(c *gin.Context) {
+	uid, ok := notifActor(c)
+	if !ok {
+		return
+	}
+	var req models.UpdateNotificationPreferencesRequest
+	if err := c.ShouldBindJSON(&req); err != nil {
+		errx.JSON(c, errx.New(errx.BadRequest, "invalid payload"))
+		return
+	}
+	if xerr := h.NotificationService.UpdatePreferences(c.Request.Context(), uid, &req.Preferences); xerr != nil {
+		errx.JSON(c, xerr)
+		return
+	}
+	c.JSON(http.StatusOK, gin.H{"preferences": req.Preferences})
+}
+
+// ListNotifications returns the caller's recent feed + the unread count.
+func (h *Handler) ListNotifications(c *gin.Context) {
+	uid, ok := notifActor(c)
+	if !ok {
+		return
+	}
+	limit := 50
+	if l := c.Query("limit"); l != "" {
+		if n, e := strconv.Atoi(l); e == nil {
+			limit = n
+		}
+	}
+	unreadOnly := c.Query("unread") == "1" || c.Query("unread") == "true"
+	items, xerr := h.NotificationService.List(c.Request.Context(), uid, limit, unreadOnly)
+	if xerr != nil {
+		errx.JSON(c, xerr)
+		return
+	}
+	unread, _ := h.NotificationService.UnreadCount(c.Request.Context(), uid)
+	c.JSON(http.StatusOK, gin.H{"notifications": items, "unread": unread})
+}
+
+// MarkNotificationRead marks one notification read.
+func (h *Handler) MarkNotificationRead(c *gin.Context) {
+	uid, ok := notifActor(c)
+	if !ok {
+		return
+	}
+	id, err := uuid.Parse(c.Param("id"))
+	if err != nil {
+		errx.JSON(c, errx.ErrUuid)
+		return
+	}
+	if xerr := h.NotificationService.MarkRead(c.Request.Context(), uid, id); xerr != nil {
+		errx.JSON(c, xerr)
+		return
+	}
+	c.JSON(http.StatusOK, gin.H{"ok": true})
+}
+
+// MarkAllNotificationsRead marks the caller's whole feed read (PUT on the
+// collection — keeps the route tree clear of a static-vs-:id conflict).
+func (h *Handler) MarkAllNotificationsRead(c *gin.Context) {
+	uid, ok := notifActor(c)
+	if !ok {
+		return
+	}
+	if xerr := h.NotificationService.MarkAllRead(c.Request.Context(), uid); xerr != nil {
+		errx.JSON(c, xerr)
+		return
+	}
+	c.JSON(http.StatusOK, gin.H{"ok": true})
+}

internal/api/handler/twofa.go

@@ -0,0 +1,93 @@
+package handler
+
+import (
+	"net/http"
+
+	"github.com/gin-gonic/gin"
+
+	"github.com/warmbly/warmbly/internal/errx"
+)
+
+// TwoFAStatus reports whether the caller has 2FA enabled.
+func (h *Handler) TwoFAStatus(c *gin.Context) {
+	uid, ok := notifActor(c)
+	if !ok {
+		return
+	}
+	enabled, err := h.TwoFAService.IsEnabled(c.Request.Context(), uid)
+	if err != nil {
+		errx.JSON(c, errx.InternalError())
+		return
+	}
+	c.JSON(http.StatusOK, gin.H{"enabled": enabled})
+}
+
+// TwoFAEnrollStart begins enrollment, returning the secret + otpauth URI once.
+func (h *Handler) TwoFAEnrollStart(c *gin.Context) {
+	uid, ok := notifActor(c)
+	if !ok {
+		return
+	}
+	res, xerr := h.TwoFAService.EnrollStart(c.Request.Context(), uid)
+	if xerr != nil {
+		errx.JSON(c, xerr)
+		return
+	}
+	c.JSON(http.StatusOK, res)
+}
+
+// TwoFAEnrollConfirm verifies a code, enables 2FA, and returns recovery codes once.
+func (h *Handler) TwoFAEnrollConfirm(c *gin.Context) {
+	uid, ok := notifActor(c)
+	if !ok {
+		return
+	}
+	var body struct {
+		Code string `json:"code"`
+	}
+	if err := c.ShouldBindJSON(&body); err != nil {
+		errx.JSON(c, errx.New(errx.BadRequest, "invalid payload"))
+		return
+	}
+	codes, xerr := h.TwoFAService.EnrollConfirm(c.Request.Context(), uid, body.Code)
+	if xerr != nil {
+		errx.JSON(c, xerr)
+		return
+	}
+	c.JSON(http.StatusOK, gin.H{"recovery_codes": codes})
+}
+
+// TwoFADisable turns off 2FA (requires a current TOTP or recovery code).
+func (h *Handler) TwoFADisable(c *gin.Context) {
+	uid, ok := notifActor(c)
+	if !ok {
+		return
+	}
+	var body struct {
+		Code string `json:"code"`
+	}
+	_ = c.ShouldBindJSON(&body)
+	if xerr := h.TwoFAService.Disable(c.Request.Context(), uid, body.Code); xerr != nil {
+		errx.JSON(c, xerr)
+		return
+	}
+	c.JSON(http.StatusOK, gin.H{"ok": true})
+}
+
+// TwoFAVerifyLogin (PUBLIC) exchanges a pending token + code for a real session.
+func (h *Handler) TwoFAVerifyLogin(c *gin.Context) {
+	var body struct {
+		PendingToken string `json:"pending_token"`
+		Code         string `json:"code"`
+	}
+	if err := c.ShouldBindJSON(&body); err != nil {
+		errx.JSON(c, errx.New(errx.BadRequest, "invalid payload"))
+		return
+	}
+	tok, xerr := h.TwoFAService.VerifyLogin(c.Request.Context(), body.PendingToken, body.Code, c.ClientIP(), c.Request.UserAgent())
+	if xerr != nil {
+		errx.JSON(c, xerr)
+		return
+	}
+	c.JSON(http.StatusOK, tok)
+}

internal/api/routes.go

@@ -159,6 +159,11 @@ func Run(
 		// resolves, and the challenge + signature are the protection.
 		auth.POST("/passkey/login/begin", h.PasskeyLoginBegin)
 		auth.POST("/passkey/login/finish", h.PasskeyLoginFinish)
+
+		// 2FA login challenge (PUBLIC): exchanges a single-use pending token +
+		// TOTP/recovery code for a real session. Rate-limited in the service
+		// (no user context here, so RateLimitMiddleware would be a no-op).
+		auth.POST("/2fa/verify", h.TwoFAVerifyLogin)
 	}
 
 	protectedAuth := auth.Group("")
@@ -179,6 +184,19 @@ func Run(
 		protectedAuth.POST("/me/avatar", h.UploadUserAvatar)
 		protectedAuth.DELETE("/me/avatar", h.DeleteUserAvatar)
 
+		// Notification preferences + in-app feed (user-scoped, no org gate).
+		protectedAuth.GET("/me/notification-preferences", h.GetNotificationPreferences)
+		protectedAuth.PUT("/me/notification-preferences", h.UpdateNotificationPreferences)
+		protectedAuth.GET("/me/notifications", h.ListNotifications)
+		protectedAuth.PUT("/me/notifications", h.MarkAllNotificationsRead)
+		protectedAuth.POST("/me/notifications/:id/read", h.MarkNotificationRead)
+
+		// 2FA enrollment + management (user-scoped, behind a live session).
+		protectedAuth.GET("/2fa/status", h.TwoFAStatus)
+		protectedAuth.POST("/2fa/enroll/start", h.TwoFAEnrollStart)
+		protectedAuth.POST("/2fa/enroll/confirm", h.TwoFAEnrollConfirm)
+		protectedAuth.DELETE("/2fa", h.TwoFADisable)
+
 		// Passkey enrollment + management require an authenticated session.
 		protectedAuth.POST("/passkey/register/begin", h.PasskeyRegisterBegin)
 		protectedAuth.POST("/passkey/register/finish", h.PasskeyRegisterFinish)