package notifications_test

import (

"bytes"

"context"

_ "embed"

"encoding/json"

"flag"

"fmt"

"go/ast"

"go/parser"

"go/token"

"io"

"net/http"

"net/http/httptest"

"net/url"

"os"

"path/filepath"

"regexp"

"slices"

"sort"

"strings"

"sync"

"testing"

"text/template"

"time"

"github.com/emersion/go-sasl"

"github.com/google/go-cmp/cmp"

"github.com/google/uuid"

smtpmock "github.com/mocktools/go-smtp-mock/v2"

"github.com/stretchr/testify/assert"

"github.com/stretchr/testify/require"

"go.uber.org/goleak"

"golang.org/x/xerrors"

"cdr.dev/slog"

"cdr.dev/slog/sloggers/sloghuman"

"github.com/coder/quartz"

"github.com/coder/serpent"

"github.com/coder/coder/v2/coderd/coderdtest"

"github.com/coder/coder/v2/coderd/database"

"github.com/coder/coder/v2/coderd/database/dbauthz"

"github.com/coder/coder/v2/coderd/database/dbgen"

"github.com/coder/coder/v2/coderd/database/dbtestutil"

"github.com/coder/coder/v2/coderd/notifications"

"github.com/coder/coder/v2/coderd/notifications/dispatch"

"github.com/coder/coder/v2/coderd/notifications/dispatch/smtptest"

"github.com/coder/coder/v2/coderd/notifications/types"

"github.com/coder/coder/v2/coderd/rbac"

"github.com/coder/coder/v2/coderd/util/syncmap"

"github.com/coder/coder/v2/codersdk"

"github.com/coder/coder/v2/testutil"

)

// updateGoldenFiles is a flag that can be set to update golden files.

var updateGoldenFiles = flag.Bool("update", false, "Update golden files")

func TestMain(m *testing.M) {

goleak.VerifyTestMain(m, testutil.GoleakOptions...)

}

// TestBasicNotificationRoundtrip enqueues a message to the store, waits for it to be acquired by a notifier,

// passes it off to a fake handler, and ensures the results are synchronized to the store.

func TestBasicNotificationRoundtrip(t *testing.T) {

t.Parallel()

// SETUP

if !dbtestutil.WillUsePostgres() {

t.Skip("This test requires postgres; it relies on business-logic only implemented in the database")

}

ctx := dbauthz.AsNotifier(testutil.Context(t, testutil.WaitSuperLong))

store, pubsub := dbtestutil.NewDB(t)

logger := testutil.Logger(t)

method := database.NotificationMethodSmtp

// GIVEN: a manager with standard config but a faked dispatch handler

handler := &fakeHandler{}

interceptor := &syncInterceptor{Store: store}

cfg := defaultNotificationsConfig(method)

cfg.RetryInterval = serpent.Duration(time.Hour) // Ensure retries don't interfere with the test

mgr, err := notifications.NewManager(cfg, interceptor, pubsub, defaultHelpers(), createMetrics(), logger.Named("manager"))

require.NoError(t, err)

mgr.WithHandlers(map[database.NotificationMethod]notifications.Handler{

method: handler,

database.NotificationMethodInbox: &fakeHandler{},

})

t.Cleanup(func() {

assert.NoError(t, mgr.Stop(ctx))

})

enq, err := notifications.NewStoreEnqueuer(cfg, store, defaultHelpers(), logger.Named("enqueuer"), quartz.NewReal())

require.NoError(t, err)

user := createSampleUser(t, store)

// WHEN: 2 messages are enqueued

sid, err := enq.Enqueue(ctx, user.ID, notifications.TemplateWorkspaceDeleted, map[string]string{"type": "success"}, "test")

require.NoError(t, err)

fid, err := enq.Enqueue(ctx, user.ID, notifications.TemplateWorkspaceDeleted, map[string]string{"type": "failure"}, "test")

require.NoError(t, err)

mgr.Run(ctx)

// THEN: we expect that the handler will have received the notifications for dispatch

require.Eventually(t, func() bool {

handler.mu.RLock()

defer handler.mu.RUnlock()

return slices.Contains(handler.succeeded, sid[0].String()) &&

slices.Contains(handler.failed, fid[0].String())

}, testutil.WaitLong, testutil.IntervalFast)

// THEN: we expect the store to be called with the updates of the earlier dispatches

require.Eventually(t, func() bool {

return interceptor.sent.Load() == 2 &&

interceptor.failed.Load() == 2

}, testutil.WaitLong, testutil.IntervalFast)

// THEN: we verify that the store contains notifications in their expected state

success, err := store.GetNotificationMessagesByStatus(ctx, database.GetNotificationMessagesByStatusParams{

Status: database.NotificationMessageStatusSent,

Limit: 10,

})

require.NoError(t, err)

require.Len(t, success, 2)

failed, err := store.GetNotificationMessagesByStatus(ctx, database.GetNotificationMessagesByStatusParams{

Status: database.NotificationMessageStatusTemporaryFailure,

Limit: 10,

})

require.NoError(t, err)

require.Len(t, failed, 2)

}

func TestSMTPDispatch(t *testing.T) {

t.Parallel()

// SETUP

ctx := dbauthz.AsNotifier(testutil.Context(t, testutil.WaitSuperLong))

store, pubsub := dbtestutil.NewDB(t)

logger := testutil.Logger(t)

// start mock SMTP server

mockSMTPSrv := smtpmock.New(smtpmock.ConfigurationAttr{

LogToStdout: false,

LogServerActivity: true,

})

require.NoError(t, mockSMTPSrv.Start())

t.Cleanup(func() {

assert.NoError(t, mockSMTPSrv.Stop())

})

// GIVEN: an SMTP setup referencing a mock SMTP server

const from = "danny@coder.com"

method := database.NotificationMethodSmtp

cfg := defaultNotificationsConfig(method)

cfg.SMTP = codersdk.NotificationsEmailConfig{

From: from,

Smarthost: serpent.String(fmt.Sprintf("localhost:%d", mockSMTPSrv.PortNumber())),

Hello: "localhost",

}

handler := newDispatchInterceptor(dispatch.NewSMTPHandler(cfg.SMTP, logger.Named("smtp")))

mgr, err := notifications.NewManager(cfg, store, pubsub, defaultHelpers(), createMetrics(), logger.Named("manager"))

require.NoError(t, err)

mgr.WithHandlers(map[database.NotificationMethod]notifications.Handler{

method: handler,

database.NotificationMethodInbox: &fakeHandler{},

})

t.Cleanup(func() {

assert.NoError(t, mgr.Stop(ctx))

})

enq, err := notifications.NewStoreEnqueuer(cfg, store, defaultHelpers(), logger.Named("enqueuer"), quartz.NewReal())

require.NoError(t, err)

user := createSampleUser(t, store)

// WHEN: a message is enqueued

msgID, err := enq.Enqueue(ctx, user.ID, notifications.TemplateWorkspaceDeleted, map[string]string{}, "test")

require.NoError(t, err)

require.Len(t, msgID, 2)

mgr.Run(ctx)

// THEN: wait until the dispatch interceptor validates that the messages were dispatched

require.Eventually(t, func() bool {

assert.Nil(t, handler.lastErr.Load())

assert.True(t, handler.retryable.Load() == 0)

return handler.sent.Load() == 1

}, testutil.WaitLong, testutil.IntervalMedium)

// THEN: we verify that the expected message was received by the mock SMTP server

msgs := mockSMTPSrv.MessagesAndPurge()

require.Len(t, msgs, 1)

require.Contains(t, msgs[0].MsgRequest(), fmt.Sprintf("From: %s", from))

require.Contains(t, msgs[0].MsgRequest(), fmt.Sprintf("To: %s", user.Email))

require.Contains(t, msgs[0].MsgRequest(), fmt.Sprintf("Message-Id: %s", msgID[0]))

}

func TestWebhookDispatch(t *testing.T) {

t.Parallel()

// SETUP

ctx := dbauthz.AsNotifier(testutil.Context(t, testutil.WaitSuperLong))

store, pubsub := dbtestutil.NewDB(t)

logger := testutil.Logger(t)

sent := make(chan dispatch.WebhookPayload, 1)

// Mock server to simulate webhook endpoint.

server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {

var payload dispatch.WebhookPayload

err := json.NewDecoder(r.Body).Decode(&payload)

assert.NoError(t, err)

assert.Equal(t, "application/json", r.Header.Get("Content-Type"))

w.WriteHeader(http.StatusOK)

_, err = w.Write([]byte("noted."))

assert.NoError(t, err)

sent <- payload

}))

defer server.Close()

endpoint, err := url.Parse(server.URL)

require.NoError(t, err)

// GIVEN: a webhook setup referencing a mock HTTP server to receive the webhook

cfg := defaultNotificationsConfig(database.NotificationMethodWebhook)

cfg.Webhook = codersdk.NotificationsWebhookConfig{

Endpoint: *serpent.URLOf(endpoint),

}

mgr, err := notifications.NewManager(cfg, store, pubsub, defaultHelpers(), createMetrics(), logger.Named("manager"))

require.NoError(t, err)

t.Cleanup(func() {

assert.NoError(t, mgr.Stop(ctx))

})

enq, err := notifications.NewStoreEnqueuer(cfg, store, defaultHelpers(), logger.Named("enqueuer"), quartz.NewReal())

require.NoError(t, err)

const (

email = "bob@coder.com"

name = "Robert McBobbington"

username = "bob"

)

user := dbgen.User(t, store, database.User{

Email: email,

Username: username,

Name: name,

})

// WHEN: a notification is enqueued (including arbitrary labels)

input := map[string]string{

"a": "b",

"c": "d",

}

msgID, err := enq.Enqueue(ctx, user.ID, notifications.TemplateWorkspaceDeleted, input, "test")

require.NoError(t, err)

mgr.Run(ctx)

// THEN: the webhook is received by the mock server and has the expected contents

payload := testutil.TryReceive(testutil.Context(t, testutil.WaitShort), t, sent)

require.EqualValues(t, "1.1", payload.Version)

require.Equal(t, msgID[0], payload.MsgID)

require.Equal(t, payload.Payload.Labels, input)

require.Equal(t, payload.Payload.UserEmail, email)

// UserName is coalesced from `name` and `username`; in this case `name` wins.

// This is not strictly necessary for this test, but it's testing some side logic which is too small for its own test.

require.Equal(t, payload.Payload.UserName, name)

require.Equal(t, payload.Payload.UserUsername, username)

// Right now we don't have a way to query notification templates by ID in dbmem, and it's not necessary to add this

// just to satisfy this test. We can safely assume that as long as this value is not empty that the given value was delivered.

require.NotEmpty(t, payload.Payload.NotificationName)

}

// TestBackpressure validates that delays in processing the buffered updates will result in slowed dequeue rates.

// As a side-effect, this also tests the graceful shutdown and flushing of the buffers.

func TestBackpressure(t *testing.T) {

t.Parallel()

// SETUP

if !dbtestutil.WillUsePostgres() {

t.Skip("This test requires postgres; it relies on business-logic only implemented in the database")

}

store, pubsub := dbtestutil.NewDB(t)

logger := testutil.Logger(t)

ctx := dbauthz.AsNotifier(testutil.Context(t, testutil.WaitShort))

const method = database.NotificationMethodWebhook

cfg := defaultNotificationsConfig(method)

// Tune the queue to fetch often.

const fetchInterval = time.Millisecond * 200

const batchSize = 10

cfg.FetchInterval = serpent.Duration(fetchInterval)

cfg.LeaseCount = serpent.Int64(batchSize)

// never time out for this test

cfg.LeasePeriod = serpent.Duration(time.Hour)

cfg.DispatchTimeout = serpent.Duration(time.Hour - time.Millisecond)

// Shrink buffers down and increase flush interval to provoke backpressure.

// Flush buffers every 5 fetch intervals.

const syncInterval = time.Second

cfg.StoreSyncInterval = serpent.Duration(syncInterval)

cfg.StoreSyncBufferSize = serpent.Int64(2)

handler := &chanHandler{calls: make(chan dispatchCall)}

// Intercept calls to submit the buffered updates to the store.

storeInterceptor := &syncInterceptor{Store: store}

mClock := quartz.NewMock(t)

syncTrap := mClock.Trap().NewTicker("Manager", "storeSync")

defer syncTrap.Close()

fetchTrap := mClock.Trap().TickerFunc("notifier", "fetchInterval")

defer fetchTrap.Close()

// GIVEN: a notification manager whose updates will be intercepted

mgr, err := notifications.NewManager(cfg, storeInterceptor, 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, defaultHelpers(), logger.Named("enqueuer"), mClock)

require.NoError(t, err)

user := createSampleUser(t, store)

// WHEN: a set of notifications are enqueued, which causes backpressure due to the batchSize which can be processed per fetch

const totalMessages = 30

for i := range totalMessages {

_, err = enq.Enqueue(ctx, user.ID, notifications.TemplateWorkspaceDeleted, map[string]string{"i": fmt.Sprintf("%d", i)}, "test")

require.NoError(t, err)

}

// Start the notifier.

mgr.Run(ctx)

syncTrap.MustWait(ctx).MustRelease(ctx)

fetchTrap.MustWait(ctx).MustRelease(ctx)

// THEN:

// Trigger a fetch

w := mClock.Advance(fetchInterval)

// one batch of dispatches is sent

for range batchSize {

call := testutil.TryReceive(ctx, t, handler.calls)

testutil.RequireSend(ctx, t, call.result, dispatchResult{

retryable: false,

err: nil,

})

}

// The first fetch will not complete, because of the short sync buffer of 2. This is the

// backpressure.

select {

case <-time.After(testutil.IntervalMedium):

// success

case <-w.Done():

t.Fatal("fetch completed despite backpressure")

}

// We expect that the store will have received NO updates.

require.EqualValues(t, 0, storeInterceptor.sent.Load()+storeInterceptor.failed.Load())

// However, when we Stop() the manager the backpressure will be relieved and the buffered updates will ALL be flushed,

// since all the goroutines that were blocked (on writing updates to the buffer) will be unblocked and will complete.

// Stop() waits for the in-progress flush to complete, meaning we have to advance the time such that sync triggers

// a total of (batchSize/StoreSyncBufferSize)-1 times. The -1 is because once we run the penultimate sync, it

// clears space in the buffer for the last dispatches of the batch, which allows graceful shutdown to continue

// immediately, without waiting for the last trigger.

stopErr := make(chan error, 1)

go func() {

stopErr <- mgr.Stop(ctx)

}()

elapsed := fetchInterval

syncEnd := time.Duration(batchSize/cfg.StoreSyncBufferSize.Value()-1) * cfg.StoreSyncInterval.Value()

t.Logf("will advance until %dms have elapsed", syncEnd.Milliseconds())

for elapsed < syncEnd {

d, wt := mClock.AdvanceNext()

elapsed += d

t.Logf("elapsed: %dms", elapsed.Milliseconds())

// fetches complete immediately, since TickerFunc only allows one call to the callback in flight at at time.

wt.MustWait(ctx)

if elapsed%cfg.StoreSyncInterval.Value() == 0 {

numSent := cfg.StoreSyncBufferSize.Value() * int64(elapsed/cfg.StoreSyncInterval.Value())

t.Logf("waiting for %d messages", numSent)

require.Eventually(t, func() bool {

// need greater or equal because the last set of messages can come immediately due

// to graceful shut down

return int64(storeInterceptor.sent.Load()) >= numSent

}, testutil.WaitShort, testutil.IntervalFast)

}

}

t.Log("done advancing")

// The batch completes

w.MustWait(ctx)

require.NoError(t, testutil.TryReceive(ctx, t, stopErr))

require.EqualValues(t, batchSize, storeInterceptor.sent.Load()+storeInterceptor.failed.Load())

}

func TestRetries(t *testing.T) {

t.Parallel()

// SETUP

if !dbtestutil.WillUsePostgres() {

t.Skip("This test requires postgres; it relies on business-logic only implemented in the database")

}

const maxAttempts = 3

ctx := dbauthz.AsNotifier(testutil.Context(t, testutil.WaitSuperLong))

store, pubsub := dbtestutil.NewDB(t)

logger := testutil.Logger(t)

// GIVEN: a mock HTTP server which will receive webhooksand a map to track the dispatch attempts

receivedMap := syncmap.New[uuid.UUID, int]()

// Mock server to simulate webhook endpoint.

server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {

var payload dispatch.WebhookPayload

err := json.NewDecoder(r.Body).Decode(&payload)

assert.NoError(t, err)

count, _ := receivedMap.LoadOrStore(payload.MsgID, 0)

count++

receivedMap.Store(payload.MsgID, count)

// Let the request succeed if this is its last attempt.

if count == maxAttempts {

w.WriteHeader(http.StatusOK)

_, err = w.Write([]byte("noted."))

assert.NoError(t, err)

return

}

w.WriteHeader(http.StatusInternalServerError)

_, err = w.Write([]byte("retry again later..."))

assert.NoError(t, err)

}))

defer server.Close()

endpoint, err := url.Parse(server.URL)

require.NoError(t, err)

method := database.NotificationMethodWebhook

cfg := defaultNotificationsConfig(method)

cfg.Webhook = codersdk.NotificationsWebhookConfig{

Endpoint: *serpent.URLOf(endpoint),

}

cfg.MaxSendAttempts = maxAttempts

// Tune intervals low to speed up test.

cfg.StoreSyncInterval = serpent.Duration(time.Millisecond * 100)

cfg.RetryInterval = serpent.Duration(time.Second) // query uses second-precision

cfg.FetchInterval = serpent.Duration(time.Millisecond * 100)

handler := newDispatchInterceptor(dispatch.NewWebhookHandler(cfg.Webhook, logger.Named("webhook")))

// Intercept calls to submit the buffered updates to the store.

storeInterceptor := &syncInterceptor{Store: store}

mgr, err := notifications.NewManager(cfg, storeInterceptor, pubsub, defaultHelpers(), createMetrics(), logger.Named("manager"))

require.NoError(t, err)

t.Cleanup(func() {

assert.NoError(t, mgr.Stop(ctx))

})

mgr.WithHandlers(map[database.NotificationMethod]notifications.Handler{

method: handler,

database.NotificationMethodInbox: &fakeHandler{},

})

enq, err := notifications.NewStoreEnqueuer(cfg, store, defaultHelpers(), logger.Named("enqueuer"), quartz.NewReal())

require.NoError(t, err)

user := createSampleUser(t, store)

// WHEN: a few notifications are enqueued, which will all fail until their final retry (determined by the mock server)

const msgCount = 5

for i := 0; i < msgCount; i++ {

_, err = enq.Enqueue(ctx, user.ID, notifications.TemplateWorkspaceDeleted, map[string]string{"i": fmt.Sprintf("%d", i)}, "test")

require.NoError(t, err)

}

mgr.Run(ctx)

// the number of tries is equal to the number of messages times the number of attempts

// times 2 as the Enqueue method pushes into both the defined dispatch method and inbox

nbTries := msgCount * maxAttempts * 2

// THEN: we expect to see all but the final attempts failing on webhook, and all messages to fail on inbox

require.Eventually(t, func() bool {

// nolint:gosec

return storeInterceptor.failed.Load() == int32(nbTries-msgCount) &&

storeInterceptor.sent.Load() == msgCount

}, testutil.WaitLong, testutil.IntervalFast)

}

// TestExpiredLeaseIsRequeued validates that notification messages which are left in "leased" status will be requeued once their lease expires.

// "leased" is the status which messages are set to when they are acquired for processing, and this should not be a terminal

// state unless the Manager shuts down ungracefully; the Manager is responsible for updating these messages' statuses once

// they have been processed.

func TestExpiredLeaseIsRequeued(t *testing.T) {

t.Parallel()

// SETUP

if !dbtestutil.WillUsePostgres() {

t.Skip("This test requires postgres; it relies on business-logic only implemented in the database")

}

ctx := dbauthz.AsNotifier(testutil.Context(t, testutil.WaitSuperLong))

store, pubsub := dbtestutil.NewDB(t)

logger := testutil.Logger(t)

// GIVEN: a manager which has its updates intercepted and paused until measurements can be taken

const (

leasePeriod = time.Second

msgCount = 5

method = database.NotificationMethodSmtp

)

cfg := defaultNotificationsConfig(method)

// Set low lease period to speed up tests.

cfg.LeasePeriod = serpent.Duration(leasePeriod)

cfg.DispatchTimeout = serpent.Duration(leasePeriod - time.Millisecond)

noopInterceptor := newNoopStoreSyncer(store)

mgrCtx, cancelManagerCtx := context.WithCancel(dbauthz.AsNotifier(context.Background()))

t.Cleanup(cancelManagerCtx)

mgr, err := notifications.NewManager(cfg, noopInterceptor, pubsub, defaultHelpers(), createMetrics(), logger.Named("manager"))

require.NoError(t, err)

enq, err := notifications.NewStoreEnqueuer(cfg, store, defaultHelpers(), logger.Named("enqueuer"), quartz.NewReal())

require.NoError(t, err)

user := createSampleUser(t, store)

// WHEN: a few notifications are enqueued which will all succeed

var msgs []string

for i := 0; i < msgCount; i++ {

ids, err := enq.Enqueue(ctx, user.ID, notifications.TemplateWorkspaceDeleted,

map[string]string{"type": "success", "index": fmt.Sprintf("%d", i)}, "test")

require.NoError(t, err)

require.Len(t, ids, 2)

msgs = append(msgs, ids[0].String(), ids[1].String())

}

mgr.Run(mgrCtx)

// THEN:

// Wait for the messages to be acquired

<-noopInterceptor.acquiredChan

// Then cancel the context, forcing the notification manager to shutdown ungracefully (simulating a crash); leaving messages in "leased" status.

cancelManagerCtx()

// Fetch any messages currently in "leased" status, and verify that they're exactly the ones we enqueued.

leased, err := store.GetNotificationMessagesByStatus(ctx, database.GetNotificationMessagesByStatusParams{

Status: database.NotificationMessageStatusLeased,

Limit: msgCount * 2,

})

require.NoError(t, err)

var leasedIDs []string

for _, msg := range leased {

leasedIDs = append(leasedIDs, msg.ID.String())

}

sort.Strings(msgs)

sort.Strings(leasedIDs)

require.EqualValues(t, msgs, leasedIDs)

// Wait out the lease period; all messages should be eligible to be re-acquired.

time.Sleep(leasePeriod + time.Millisecond)

// Start a new notification manager.

// Intercept calls to submit the buffered updates to the store.

storeInterceptor := &syncInterceptor{Store: store}

handler := newDispatchInterceptor(&fakeHandler{})

mgr, err = notifications.NewManager(cfg, storeInterceptor, pubsub, defaultHelpers(), createMetrics(), logger.Named("manager"))

require.NoError(t, err)

mgr.WithHandlers(map[database.NotificationMethod]notifications.Handler{

method: handler,

database.NotificationMethodInbox: &fakeHandler{},

})

// Use regular context now.

t.Cleanup(func() {

assert.NoError(t, mgr.Stop(ctx))

})

mgr.Run(ctx)

// Wait until all messages are sent & updates flushed to the database.

require.Eventually(t, func() bool {

return handler.sent.Load() == msgCount &&

storeInterceptor.sent.Load() == msgCount*2

}, testutil.WaitLong, testutil.IntervalFast)

// Validate that no more messages are in "leased" status.

leased, err = store.GetNotificationMessagesByStatus(ctx, database.GetNotificationMessagesByStatusParams{

Status: database.NotificationMessageStatusLeased,

Limit: msgCount,

})

require.NoError(t, err)

require.Len(t, leased, 0)

}

// TestInvalidConfig validates that misconfigurations lead to errors.

func TestInvalidConfig(t *testing.T) {

t.Parallel()

store, pubsub := dbtestutil.NewDB(t)

logger := testutil.Logger(t)

// GIVEN: invalid config with dispatch period <= lease period

const (

leasePeriod = time.Second

method = database.NotificationMethodSmtp

)

cfg := defaultNotificationsConfig(method)

cfg.LeasePeriod = serpent.Duration(leasePeriod)

cfg.DispatchTimeout = serpent.Duration(leasePeriod)

// WHEN: the manager is created with invalid config

_, err := notifications.NewManager(cfg, store, pubsub, defaultHelpers(), createMetrics(), logger.Named("manager"))

// THEN: the manager will fail to be created, citing invalid config as error

require.ErrorIs(t, err, notifications.ErrInvalidDispatchTimeout)

}

func TestNotifierPaused(t *testing.T) {

t.Parallel()

// Setup.

ctx := dbauthz.AsNotifier(testutil.Context(t, testutil.WaitSuperLong))

store, pubsub := dbtestutil.NewDB(t)

logger := testutil.Logger(t)

// Prepare the test.

handler := &fakeHandler{}

method := database.NotificationMethodSmtp

user := createSampleUser(t, store)

const fetchInterval = time.Millisecond * 100

cfg := defaultNotificationsConfig(method)

cfg.FetchInterval = serpent.Duration(fetchInterval)

mgr, err := notifications.NewManager(cfg, store, pubsub, defaultHelpers(), createMetrics(), logger.Named("manager"))

require.NoError(t, err)

mgr.WithHandlers(map[database.NotificationMethod]notifications.Handler{

method: handler,

database.NotificationMethodInbox: &fakeHandler{},

})

t.Cleanup(func() {

assert.NoError(t, mgr.Stop(ctx))

})

enq, err := notifications.NewStoreEnqueuer(cfg, store, defaultHelpers(), logger.Named("enqueuer"), quartz.NewReal())

require.NoError(t, err)

// Pause the notifier.

settingsJSON, err := json.Marshal(&codersdk.NotificationsSettings{NotifierPaused: true})

require.NoError(t, err)

err = store.UpsertNotificationsSettings(ctx, string(settingsJSON))

require.NoError(t, err)

// Start the manager so that notifications are processed, except it will be paused at this point.

// If it is started before pausing, there's a TOCTOU possibility between checking whether the notifier is paused or

// not, and processing the messages (see notifier.run).

mgr.Run(ctx)

// Notifier is paused, enqueue the next message.

sid, err := enq.Enqueue(ctx, user.ID, notifications.TemplateWorkspaceDeleted, map[string]string{"type": "success", "i": "1"}, "test")

require.NoError(t, err)

// Ensure we have a pending message and it's the expected one.

pendingMessages, err := store.GetNotificationMessagesByStatus(ctx, database.GetNotificationMessagesByStatusParams{

Status: database.NotificationMessageStatusPending,

Limit: 10,

})

require.NoError(t, err)

require.Len(t, pendingMessages, 2)

require.Equal(t, pendingMessages[0].ID.String(), sid[0].String())

require.Equal(t, pendingMessages[1].ID.String(), sid[1].String())

// Wait a few fetch intervals to be sure that no new notifications are being sent.

// TODO: use quartz instead.

// nolint:gocritic // These magic numbers are fine.

require.Eventually(t, func() bool {

handler.mu.RLock()

defer handler.mu.RUnlock()

return len(handler.succeeded)+len(handler.failed) == 0

}, fetchInterval*5, testutil.IntervalFast)

// Unpause the notifier.

settingsJSON, err = json.Marshal(&codersdk.NotificationsSettings{NotifierPaused: false})

require.NoError(t, err)

err = store.UpsertNotificationsSettings(ctx, string(settingsJSON))

require.NoError(t, err)

// Notifier is running again, message should be dequeued.

// nolint:gocritic // These magic numbers are fine.

require.Eventually(t, func() bool {

handler.mu.RLock()

defer handler.mu.RUnlock()

return slices.Contains(handler.succeeded, sid[0].String())

}, fetchInterval*5, testutil.IntervalFast)

}

//go:embed events.go

var events []byte

// enumerateAllTemplates gets all the template names from the coderd/notifications/events.go file.

// TODO(dannyk): use code-generation to create a list of all templates: https://github.com/coder/team-coconut/issues/36

func enumerateAllTemplates(t *testing.T) ([]string, error) {

t.Helper()

fset := token.NewFileSet()

node, err := parser.ParseFile(fset, "", bytes.NewBuffer(events), parser.AllErrors)

if err != nil {

return nil, err

}

var out []string

// Traverse the AST and extract variable names.

ast.Inspect(node, func(n ast.Node) bool {

// Check if the node is a declaration statement.

if decl, ok := n.(*ast.GenDecl); ok && decl.Tok == token.VAR {

for _, spec := range decl.Specs {

// Type assert the spec to a ValueSpec.

if valueSpec, ok := spec.(*ast.ValueSpec); ok {

for _, name := range valueSpec.Names {

out = append(out, name.String())

}

}

}

}

return true

})

return out, nil

}

func TestNotificationTemplates_Golden(t *testing.T) {

t.Parallel()

if !dbtestutil.WillUsePostgres() {

t.Skip("This test requires postgres; it relies on the notification templates added by migrations in the database")

}

const (

username = "bob"

password = "🤫"

hello = "localhost"

from = "system@coder.com"

hint = "run \"make gen/golden-files\" and commit the changes"

)

tests := []struct {

name string

id uuid.UUID

payload types.MessagePayload

appName string

logoURL string

}{

{

name: "TemplateWorkspaceDeleted",

id: notifications.TemplateWorkspaceDeleted,

payload: types.MessagePayload{

UserName: "Bobby",

UserEmail: "bobby@coder.com",

UserUsername: "bobby",

Labels: map[string]string{

"name": "bobby-workspace",

"reason": "autodeleted due to dormancy",

"initiator": "autobuild",

},

Targets: []uuid.UUID{

uuid.MustParse("5c6ea841-ca63-46cc-9c37-78734c7a788b"),

uuid.MustParse("b8355e3a-f3c5-4dd1-b382-7eb1fae7db52"),

},

},

},

{

name: "TemplateWorkspaceAutobuildFailed",

id: notifications.TemplateWorkspaceAutobuildFailed,

payload: types.MessagePayload{

UserName: "Bobby",

UserEmail: "bobby@coder.com",

UserUsername: "bobby",

Labels: map[string]string{

"name": "bobby-workspace",

"reason": "autostart",

},

Targets: []uuid.UUID{

uuid.MustParse("5c6ea841-ca63-46cc-9c37-78734c7a788b"),

uuid.MustParse("b8355e3a-f3c5-4dd1-b382-7eb1fae7db52"),

},

},

},

{

name: "TemplateWorkspaceDormant",

id: notifications.TemplateWorkspaceDormant,

payload: types.MessagePayload{

UserName: "Bobby",

UserEmail: "bobby@coder.com",

UserUsername: "bobby",

Labels: map[string]string{

"name": "bobby-workspace",

"reason": "breached the template's threshold for inactivity",

"initiator": "autobuild",

"dormancyHours": "24",

"timeTilDormant": "24 hours",

},

},

},

{

name: "TemplateWorkspaceAutoUpdated",

id: notifications.TemplateWorkspaceAutoUpdated,

payload: types.MessagePayload{

UserName: "Bobby",

UserEmail: "bobby@coder.com",

UserUsername: "bobby",

Labels: map[string]string{

"name": "bobby-workspace",

"template_version_name": "1.0",

"template_version_message": "template now includes catnip",

},

},

},

{

name: "TemplateWorkspaceMarkedForDeletion",

id: notifications.TemplateWorkspaceMarkedForDeletion,

payload: types.MessagePayload{

UserName: "Bobby",

UserEmail: "bobby@coder.com",

UserUsername: "bobby",

Labels: map[string]string{

"name": "bobby-workspace",

"reason": "template updated to new dormancy policy",

"dormancyHours": "24",

"timeTilDormant": "24 hours",

},

},

},

{

name: "TemplateUserAccountCreated",

id: notifications.TemplateUserAccountCreated,

payload: types.MessagePayload{

UserName: "Bobby",

UserEmail: "bobby@coder.com",

UserUsername: "bobby",

Labels: map[string]string{

"created_account_name": "bobby",

"created_account_user_name": "William Tables",

"initiator": "rob",

},

},

},

{

name: "TemplateUserAccountDeleted",

id: notifications.TemplateUserAccountDeleted,

payload: types.MessagePayload{

UserName: "Bobby",

UserEmail: "bobby@coder.com",

UserUsername: "bobby",

Labels: map[string]string{

"deleted_account_name": "bobby",

"deleted_account_user_name": "William Tables",

"initiator": "rob",

},

},

},

{

name: "TemplateUserAccountSuspended",

id: notifications.TemplateUserAccountSuspended,

payload: types.MessagePayload{

UserName: "Bobby",

UserEmail: "bobby@coder.com",

UserUsername: "bobby",

Labels: map[string]string{

"suspended_account_name": "bobby",

"suspended_account_user_name": "William Tables",

"initiator": "rob",

},

},

},

{

name: "TemplateUserAccountActivated",

id: notifications.TemplateUserAccountActivated,

payload: types.MessagePayload{

UserName: "Bobby",

UserEmail: "bobby@coder.com",

UserUsername: "bobby",

Labels: map[string]string{

"activated_account_name": "bobby",

"activated_account_user_name": "William Tables",

"initiator": "rob",

},

},

},

{

name: "TemplateYourAccountSuspended",

id: notifications.TemplateYourAccountSuspended,

payload: types.MessagePayload{

UserName: "Bobby",

UserEmail: "bobby@coder.com",

UserUsername: "bobby",

Labels: map[string]string{

"suspended_account_name": "bobby",

"initiator": "rob",

},

},

},

{

name: "TemplateYourAccountActivated",

id: notifications.TemplateYourAccountActivated,

payload: types.MessagePayload{

UserName: "Bobby",

UserEmail: "bobby@coder.com",

UserUsername: "bobby",

Labels: map[string]string{

"activated_account_name": "bobby",

"initiator": "rob",

},

},

},

{

name: "TemplateTemplateDeleted",

id: notifications.TemplateTemplateDeleted,

payload: types.MessagePayload{

UserName: "Bobby",

UserEmail: "bobby@coder.com",

UserUsername: "bobby",

Labels: map[string]string{

"name": "Bobby's Template",

"initiator": "rob",

},

},

},

{

name: "TemplateWorkspaceManualBuildFailed",

id: notifications.TemplateWorkspaceManualBuildFailed,

payload: types.MessagePayload{

UserName: "Bobby",

UserEmail: "bobby@coder.com",

UserUsername: "bobby",

Labels: map[string]string{

"name": "bobby-workspace",

"template_name": "bobby-template",

"template_version_name": "bobby-template-version",

"initiator": "joe",

"workspace_owner_username": "mrbobby",

"workspace_build_number": "3",

},

},

},

{

name: "TemplateWorkspaceBuildsFailedReport",

id: notifications.TemplateWorkspaceBuildsFailedReport,

payload: types.MessagePayload{

UserName: "Bobby",

UserEmail: "bobby@coder.com",

UserUsername: "bobby",

Labels: map[string]string{},

// We need to use floats as `json.Unmarshal` unmarshal numbers in `map[string]any` to floats.

Data: map[string]any{

"report_frequency": "week",

"templates": []map[string]any{

{

"name": "bobby-first-template",

"display_name": "Bobby First Template",

"failed_builds": 4.0,

"total_builds": 55.0,

"versions": []map[string]any{

{

"template_version_name": "bobby-template-version-1",

"failed_count": 3.0,

"failed_builds": []map[string]any{

{

"workspace_owner_username": "mtojek",

"workspace_name": "workspace-1",

"workspace_id": "24f5bd8f-1566-4374-9734-c3efa0454dc7",

"build_number": 1234.0,

},

{

"workspace_owner_username": "johndoe",

"workspace_name": "my-workspace-3",

"workspace_id": "372a194b-dcde-43f1-b7cf-8a2f3d3114a0",

"build_number": 5678.0,