start adding tests for pubsub notif

johnstcn · johnstcn · commit eeedb449d938 · 2025-04-17T16:07:24.000+01:00
diff --git a/coderd/notifications/notifications_test.go b/coderd/notifications/notifications_test.go
@@ -2088,6 +2088,64 @@ func TestNotificationOneTimePasswordDeliveryTargets(t *testing.T) {
 	})
 }
 
+func TestNotificationEnqueuePubsubNotify(t *testing.T) {
+	t.Parallel()
+	if !dbtestutil.WillUsePostgres() {
+		t.Skip("This test requires postgres; it relies on business-logic only implemented in the database")
+	}
+
+	// Given: we are between notify dispatch intervals.
+	store, pubsub := dbtestutil.NewDB(t)
+	logger := testutil.Logger(t)
+	// nolint:gocritic // Unit test.
+	ctx := dbauthz.AsNotifier(testutil.Context(t, testutil.WaitShort))
+
+	const method = database.NotificationMethodWebhook
+	cfg := defaultNotificationsConfig(method)
+
+	// Tune the queue to fetch infrequently.
+	const fetchInterval = time.Minute
+	cfg.FetchInterval = serpent.Duration(fetchInterval)
+
+	handler := &chanHandler{calls: make(chan dispatchCall)}
+
+	mClock := quartz.NewMock(t)
+	fetchTrap := mClock.Trap().TickerFunc("notifier", "fetchInterval")
+	defer fetchTrap.Close()
+
+	mgr, err := notifications.NewManager(cfg, store, pubsub, defaultHelpers(), createMetrics(),
+		logger.Named("manager"), notifications.WithTestClock(mClock))
+	require.NoError(t, err)
+	mgr.WithHandlers(map[database.NotificationMethod]notifications.Handler{
+		method:                           handler,
+		database.NotificationMethodInbox: handler,
+	})
+	enq, err := notifications.NewStoreEnqueuer(cfg, store, pubsub, defaultHelpers(), logger.Named("enqueuer"), mClock)
+	require.NoError(t, err)
+
+	user := createSampleUser(t, store)
+
+	// When: a notification is enqueued
+
+	// Then: we attempt to dispatch the notification immediately.
+	mgr.Run(ctx)
+	wt := fetchTrap.MustWait(ctx)
+	// TODO: this happens *before* manager starts listening. Need to wait.
+	_, err = enq.Enqueue(ctx, user.ID, notifications.TemplateWorkspaceDeleted, map[string]string{}, "test")
+	require.NoError(t, err)
+	wt.Release()
+	d, w := mClock.AdvanceNext()
+	t.Logf("advanced %s", d)
+
+	call := testutil.TryReceive(ctx, t, handler.calls)
+	testutil.RequireSend(ctx, t, call.result, dispatchResult{
+		retryable: false,
+		err:       nil,
+	})
+
+	<-w.Done() // TODO: this hangs and fails.
+}
+
 type fakeHandler struct {
 	mu                sync.RWMutex
 	succeeded, failed []string
diff --git a/coderd/notifications/notifier.go b/coderd/notifications/notifier.go
@@ -151,6 +151,7 @@ func (n *notifier) run(success chan<- dispatchResult, failure chan<- dispatchRes
 
 	// Also signal the processing loop when a notification is enqueued.
 	if stopListen, err := n.ps.Subscribe(EventNotificationEnqueued, func(ctx context.Context, _ []byte) {
+		n.log.Debug(n.outerCtx, "got pubsub event", slog.F("event", EventNotificationEnqueued))
 		c := make(chan struct{})
 		select {
 		case <-ctx.Done():
@@ -170,10 +171,9 @@ func (n *notifier) run(success chan<- dispatchResult, failure chan<- dispatchRes
 		defer stopListen()
 	}
 
-	// Note the order of operations here.
-	_ = tick.Wait() // will block until gracefulCtx is done
-	close(loopTick) // happens immediately
-	<-loopDone      // wait for the current processing loop to finish
+	_ = tick.Wait() // Block until the ticker exits. This will be after gracefulCtx is canceled.
+	close(loopTick) // Signal the processing goroutine to stop.
+	<-loopDone      // Wait for the processing goroutine to exit.
 
 	// only errors we can return are context errors.  Only return an error if the outer context
 	// was canceled, not if we were gracefully stopped.
