feat: add killswitch for notifications (#13794)

coderd/notifications/manager_test.go

@@ -12,13 +12,14 @@ import (
 	"github.com/stretchr/testify/require"
 	"golang.org/x/xerrors"
 
+	"github.com/coder/serpent"
+
 	"github.com/coder/coder/v2/coderd/database"
 	"github.com/coder/coder/v2/coderd/database/dbgen"
 	"github.com/coder/coder/v2/coderd/notifications"
 	"github.com/coder/coder/v2/coderd/notifications/dispatch"
 	"github.com/coder/coder/v2/coderd/notifications/types"
 	"github.com/coder/coder/v2/testutil"
-	"github.com/coder/serpent"
 )
 
 func TestBufferedUpdates(t *testing.T) {

coderd/notifications/notifications_test.go

@@ -538,6 +538,71 @@ func TestInvalidConfig(t *testing.T) {
 	require.ErrorIs(t, err, notifications.ErrInvalidDispatchTimeout)
 }
 
+func TestNotifierPaused(t *testing.T) {
+	t.Parallel()
+
+	// setup
+	ctx, logger, db := setupInMemory(t)
+
+	// Prepare the test
+	handler := &fakeHandler{}
+	method := database.NotificationMethodSmtp
+	user := createSampleUser(t, db)
+
+	cfg := defaultNotificationsConfig(method)
+	fetchInterval := time.Nanosecond // Let
+	cfg.FetchInterval = *serpent.DurationOf(&fetchInterval)
+	mgr, err := notifications.NewManager(cfg, db, logger.Named("manager"))
+	require.NoError(t, err)
+	mgr.WithHandlers(map[database.NotificationMethod]notifications.Handler{method: handler})
+	t.Cleanup(func() {
+		assert.NoError(t, mgr.Stop(ctx))
+	})
+	enq, err := notifications.NewStoreEnqueuer(cfg, db, defaultHelpers(), logger.Named("enqueuer"))
+	require.NoError(t, err)
+
+	mgr.Run(ctx)
+
+	// Notifier is on, enqueue the first message.
+	sid, err := enq.Enqueue(ctx, user.ID, notifications.TemplateWorkspaceDeleted, map[string]string{"type": "success"}, "test")
+	require.NoError(t, err)
+	require.Eventually(t, func() bool {
+		handler.mu.RLock()
+		defer handler.mu.RUnlock()
+		return slices.Contains(handler.succeeded, sid.String())
+	}, testutil.WaitShort, testutil.IntervalFast)
+
+	// Pause the notifier.
+	settingsJSON, err := json.Marshal(&codersdk.NotificationsSettings{NotifierPaused: true})
+	require.NoError(t, err)
+	err = db.UpsertNotificationsSettings(ctx, string(settingsJSON))
+	require.NoError(t, err)
+
+	// Notifier is paused, enqueue the next message.
+	sid, err = enq.Enqueue(ctx, user.ID, notifications.TemplateWorkspaceDeleted, map[string]string{"type": "success"}, "test")
+	require.NoError(t, err)
+	require.Eventually(t, func() bool {
+		pendingMessages, err := db.GetNotificationMessagesByStatus(ctx, database.GetNotificationMessagesByStatusParams{
+			Status: database.NotificationMessageStatusPending,
+		})
+		assert.NoError(t, err)
+		return len(pendingMessages) == 1
+	}, testutil.WaitShort, testutil.IntervalFast)
+
+	// Unpause the notifier.
+	settingsJSON, err = json.Marshal(&codersdk.NotificationsSettings{NotifierPaused: false})
+	require.NoError(t, err)
+	err = db.UpsertNotificationsSettings(ctx, string(settingsJSON))
+	require.NoError(t, err)
+
+	// Notifier is running again, message should be dequeued.
+	require.Eventually(t, func() bool {
+		handler.mu.RLock()
+		defer handler.mu.RUnlock()
+		return slices.Contains(handler.succeeded, sid.String())
+	}, testutil.WaitShort, testutil.IntervalFast)
+}
+
 type fakeHandler struct {
 	mu sync.RWMutex
 
@@ -546,7 +611,7 @@ type fakeHandler struct {
 }
 
 func (f *fakeHandler) Dispatcher(payload types.MessagePayload, _, _ string) (dispatch.DeliveryFunc, error) {
-	return func(ctx context.Context, msgID uuid.UUID) (retryable bool, err error) {
+	return func(_ context.Context, msgID uuid.UUID) (retryable bool, err error) {
 		f.mu.Lock()
 		defer f.mu.Unlock()
 

coderd/notifications/notifier.go

@@ -71,10 +71,18 @@ func (n *notifier) run(ctx context.Context, success chan<- dispatchResult, failu
 		default:
 		}
 
-		// Call process() immediately (i.e. don't wait an initial tick).
-		err := n.process(ctx, success, failure)
+		// Check if notifier is not paused.
+		ok, err := n.ensureRunning(ctx)
 		if err != nil {
-			n.log.Error(ctx, "failed to process messages", slog.Error(err))
+			n.log.Warn(ctx, "failed to check notifier state", slog.Error(err))
+		}
+
+		if ok {
+			// Call process() immediately (i.e. don't wait an initial tick).
+			err = n.process(ctx, success, failure)
+			if err != nil {
+				n.log.Error(ctx, "failed to process messages", slog.Error(err))
+			}
 		}
 
 		// Shortcut to bail out quickly if stop() has been called or the context canceled.
@@ -89,6 +97,31 @@ func (n *notifier) run(ctx context.Context, success chan<- dispatchResult, failu
 	}
 }
 
+// ensureRunning checks if notifier is not paused.
+func (n *notifier) ensureRunning(ctx context.Context) (bool, error) {
+	n.log.Debug(ctx, "check if notifier is paused")
+
+	settingsJSON, err := n.store.GetNotificationsSettings(ctx)
+	if err != nil {
+		return false, xerrors.Errorf("get notifications settings: %w", err)
+	}
+
+	var settings codersdk.NotificationsSettings
+	if len(settingsJSON) == 0 {
+		return true, nil // settings.NotifierPaused is false by default
+	}
+
+	err = json.Unmarshal([]byte(settingsJSON), &settings)
+	if err != nil {
+		return false, xerrors.Errorf("unmarshal notifications settings")
+	}
+
+	if settings.NotifierPaused {
+		n.log.Debug(ctx, "notifier is paused, notifications will not be delivered")
+	}
+	return !settings.NotifierPaused, nil
+}
+
 // process is responsible for coordinating the retrieval, processing, and delivery of messages.
 // Messages are dispatched concurrently, but they may block when success/failure channels are full.
 //

coderd/notifications/spec.go

@@ -21,6 +21,7 @@ type Store interface {
 	EnqueueNotificationMessage(ctx context.Context, arg database.EnqueueNotificationMessageParams) (database.NotificationMessage, error)
 	FetchNewMessageMetadata(ctx context.Context, arg database.FetchNewMessageMetadataParams) (database.FetchNewMessageMetadataRow, error)
 	GetNotificationMessagesByStatus(ctx context.Context, arg database.GetNotificationMessagesByStatusParams) ([]database.NotificationMessage, error)
+	GetNotificationsSettings(ctx context.Context) (string, error)
 }
 
 // Handler is responsible for preparing and delivering a notification by a given method.