address some but not all PR comments

johnstcn · johnstcn · commit f5ec4192ae1b · 2025-04-23T17:52:40.000+01:00
diff --git a/coderd/notifications/enqueuer.go b/coderd/notifications/enqueuer.go
@@ -85,13 +85,6 @@ func (s *StoreEnqueuer) Enqueue(ctx context.Context, userID, templateID uuid.UUI
 // Enqueue queues a notification message for later delivery.
 // Messages will be dequeued by a notifier later and dispatched.
 func (s *StoreEnqueuer) EnqueueWithData(ctx context.Context, userID, templateID uuid.UUID, labels map[string]string, data map[string]any, createdBy string, targets ...uuid.UUID) ([]uuid.UUID, error) {
-	defer func() {
-		// Publish an event to notify that a notification has been enqueued.
-		// Failure to publish is acceptable, as the fetcher will still process the
-		// message on its next run.
-		// TODO(Cian): debounce this to maybe once per second or so?
-		_ = s.ps.Publish(EventNotificationEnqueued, nil)
-	}()
 	metadata, err := s.store.FetchNewMessageMetadata(ctx, database.FetchNewMessageMetadataParams{
 		UserID:                 userID,
 		NotificationTemplateID: templateID,
@@ -171,6 +164,11 @@ func (s *StoreEnqueuer) EnqueueWithData(ctx context.Context, userID, templateID
 	}
 
 	s.log.Debug(ctx, "enqueued notification", slog.F("msg_ids", uuids))
+	// Publish an event to notify that a notification has been enqueued.
+	// Failure to publish is acceptable, as the fetcher will still process the
+	// message on its next run.
+	// TODO(Cian): debounce this to maybe once per second or so?
+	_ = s.ps.Publish(EventNotificationEnqueued, nil)
 	return uuids, nil
 }
 
diff --git a/coderd/notifications/notifier.go b/coderd/notifications/notifier.go
@@ -123,12 +123,14 @@ func (n *notifier) run(success chan<- dispatchResult, failure chan<- dispatchRes
 			if err != nil {
 				n.log.Warn(n.outerCtx, "failed to check notifier state", slog.Error(err))
 			}
+			if !ok { // Notifier is paused, skip processing.
+				close(c)
+				continue
+			}
 
-			if ok {
-				err = n.process(n.outerCtx, success, failure)
-				if err != nil {
-					n.log.Error(n.outerCtx, "failed to process messages", slog.Error(err))
-				}
+			err = n.process(n.outerCtx, success, failure)
+			if err != nil {
+				n.log.Error(n.outerCtx, "failed to process messages", slog.Error(err))
 			}
 			// Signal that we've finished processing one iteration.
 			close(c)
@@ -141,6 +143,7 @@ func (n *notifier) run(success chan<- dispatchResult, failure chan<- dispatchRes
 
 	// Periodically trigger the processing loop.
 	tick := n.clock.TickerFunc(n.gracefulCtx, n.cfg.FetchInterval.Value(), func() error {
+		// Reset the enqueue counter after each tick.
 		defer enqueueEventsThisLoop.Store(0)
 		c := make(chan struct{})
 		loopTick <- c
@@ -154,8 +157,9 @@ func (n *notifier) run(success chan<- dispatchResult, failure chan<- dispatchRes
 	if stopListen, err := n.ps.Subscribe(EventNotificationEnqueued, func(ctx context.Context, _ []byte) {
 		enqueued := enqueueEventsThisLoop.Add(1)
 		skipEarlyDispatch := enqueued > 1
-		n.log.Debug(n.outerCtx, "got pubsub event", slog.F("count", enqueued), slog.F("skip_early_dispatch", skipEarlyDispatch), slog.F("event", EventNotificationEnqueued))
-		if enqueued > 1 {
+		n.log.Debug(n.outerCtx, "TODO REMOVE THIS got pubsub event", slog.F("count", enqueued), slog.F("skip_early_dispatch", skipEarlyDispatch), slog.F("event", EventNotificationEnqueued))
+		if skipEarlyDispatch {
+			// Avoid overloading the database. We will get to these in the next tick.
 			return
 		}
 		c := make(chan struct{})
