Jul 11, 2024 · Jul 3, 2024 · Jul 5, 2024 · Jul 5, 2024 · Jul 8, 2024 · Jul 8, 2024
diff --git a/cli/server.go b/cli/server.go
 )
 if experiments.Enabled(codersdk.ExperimentNotifications) {
 cfg := options.DeploymentValues.Notifications
 metrics := notifications.NewMetrics(options.PrometheusRegistry)

 // The enqueuer is responsible for enqueueing notifications to the given store.
 enqueuer, err := notifications.NewStoreEnqueuer(cfg, options.Database, templateHelpers(options), logger.Named("notifications.enqueuer"))
 // The notification manager is responsible for:
 //   - creating notifiers and managing their lifecycles (notifiers are responsible for dequeueing/sending notifications)
 //   - keeping the store updated with status updates
 notificationsManager, err = notifications.NewManager(cfg, options.Database, logger.Named("notifications.manager"))
 notificationsManager, err = notifications.NewManager(cfg, options.Database,metrics,logger.Named("notifications.manager"))
 if err != nil {
 return xerrors.Errorf("failed to instantiate notification manager: %w", err)
 }
diff --git a/coderd/database/dbauthz/dbauthz.go b/coderd/database/dbauthz/dbauthz.go
 return q.db.DeleteWorkspaceAgentPortSharesByTemplate(ctx, templateID)
 }

 func (q *querier) EnqueueNotificationMessage(ctx context.Context, arg database.EnqueueNotificationMessageParams)(database.NotificationMessage,error) {
 func (q *querier) EnqueueNotificationMessage(ctx context.Context, arg database.EnqueueNotificationMessageParams) error {
 if err := q.authorizeContext(ctx, policy.ActionCreate, rbac.ResourceSystem); err != nil {
 returndatabase.NotificationMessage{},err
 return err
 }
 return q.db.EnqueueNotificationMessage(ctx, arg)
 }
diff --git a/coderd/database/dbmem/dbmem.go b/coderd/database/dbmem/dbmem.go
 q.mutex.Lock()
 defer q.mutex.Unlock()

 var out []database.AcquireNotificationMessagesRow
 for _, nm:=rangeq.notificationMessages {
 iflen(out) >= int(arg.Count) {
 break
 }
 // Shift the first "Count" notifications off the slice (FIFO).
 sz:=len(q.notificationMessages)
 ifsz > int(arg.Count) {
 sz = int(arg.Count)
 }

 list := q.notificationMessages[:sz]
 q.notificationMessages = q.notificationMessages[sz:]

 var out []database.AcquireNotificationMessagesRow
 for _, nm := range list {
 acquirableStatuses := []database.NotificationMessageStatus{database.NotificationMessageStatusPending, database.NotificationMessageStatusTemporaryFailure}
 if !slices.Contains(acquirableStatuses, nm.Status) {
 continue
 ID:            nm.ID,
 Payload:       nm.Payload,
 Method:        nm.Method,
 CreatedBy:     nm.CreatedBy,
 TitleTemplate: "This is a title with {{.Labels.variable}}",
 BodyTemplate:  "This is a body with {{.Labels.variable}}",
 TemplateID:    nm.NotificationTemplateID,
 })
 }

 return nil
 }

 func (q *FakeQuerier) EnqueueNotificationMessage(_ context.Context, arg database.EnqueueNotificationMessageParams)(database.NotificationMessage,error) {
 func (q *FakeQuerier) EnqueueNotificationMessage(_ context.Context, arg database.EnqueueNotificationMessageParams) error {
 err := validateDatabaseType(arg)
 if err != nil {
 returndatabase.NotificationMessage{},err
 return err
 }

 q.mutex.Lock()
 var payload types.MessagePayload
 err = json.Unmarshal(arg.Payload, &payload)
 if err != nil {
 returndatabase.NotificationMessage{},err
 return err
 }

 nm := database.NotificationMessage{

 q.notificationMessages = append(q.notificationMessages, nm)

 returnnm,err
 return err
 }

 func (q *FakeQuerier) FavoriteWorkspace(_ context.Context, arg uuid.UUID) error {
diff --git a/coderd/database/dbmetrics/dbmetrics.go b/coderd/database/dbmetrics/dbmetrics.go
diff --git a/coderd/database/dbmock/dbmock.go b/coderd/database/dbmock/dbmock.go
diff --git a/coderd/database/dump.sql b/coderd/database/dump.sql
diff --git a/coderd/database/migrations/000225_notifications_metrics.down.sql b/coderd/database/migrations/000225_notifications_metrics.down.sql
 ALTER TABLE notification_messages
 DROP COLUMN IF EXISTS queued_seconds;
diff --git a/coderd/database/migrations/000225_notifications_metrics.up.sql b/coderd/database/migrations/000225_notifications_metrics.up.sql
 ALTER TABLE notification_messages
 ADD COLUMN queued_seconds FLOAT NULL;
diff --git a/coderd/database/models.go b/coderd/database/models.go
diff --git a/coderd/database/querier.go b/coderd/database/querier.go
diff --git a/coderd/database/queries.sql.go b/coderd/database/queries.sql.go
diff --git a/coderd/database/queries/notifications.sql b/coderd/database/queries/notifications.sql
 WHERE nt.id = @notification_template_id
  AND u.id = @user_id;

 -- name: EnqueueNotificationMessage :one
 -- name: EnqueueNotificationMessage :exec
 INSERT INTO notification_messages (id, notification_template_id, user_id, method, payload, targets, created_by)
 VALUES (@id,
        @notification_template_id,
        @user_id,
        @method::notification_method,
        @payload::jsonb,
        @targets,
        @created_by)
 RETURNING *;
        @created_by);

 -- Acquires the lease for a given count of notification messages, to enable concurrent dequeuing and subsequent sending.
 -- Only rows that aren't already leased (or ones which are leased but have exceeded their lease period) are returned.
 WITH acquired AS (
    UPDATE
        notification_messages
            SET updated_at = NOW(),
            SET queued_seconds = GREATEST(0, EXTRACT(EPOCH FROM (NOW() - updated_at)))::FLOAT,
                updated_at = NOW(),
                status = 'leased'::notification_message_status,
                status_reason = 'Leased by notifier ' || sqlc.arg('notifier_id')::uuid,
                leased_until = NOW() + CONCAT(sqlc.arg('lease_seconds')::int, ' seconds')::interval
    nm.id,
    nm.payload,
    nm.method,
    nm.created_by,
    nm.attempt_count::int AS attempt_count,
    nm.queued_seconds::float AS queued_seconds,
    -- template
    nt.id AS template_id,
    nt.title_template,
    nt.body_template
 FROM acquired nm
         JOIN notification_templates nt ON nm.notification_template_id = nt.id;

 -- name: BulkMarkNotificationMessagesFailed :execrows
 UPDATE notification_messages
 SET updated_at       = subquery.failed_at,
 SET queued_seconds   = 0,
    updated_at       = subquery.failed_at,
    attempt_count    = attempt_count + 1,
    status           = CASE
                           WHEN attempt_count + 1 < @max_attempts::int THEN subquery.status

 -- name: BulkMarkNotificationMessagesSent :execrows
 UPDATE notification_messages
 SET updated_at       = new_values.sent_at,
 SET queued_seconds   = 0,
    updated_at       = new_values.sent_at,
    attempt_count    = attempt_count + 1,
    status           = 'sent'::notification_message_status,
    status_reason    = NULL,
    leased_until     = NULL,
    next_retry_after = NULL
 FROM (SELECT UNNEST(@ids::uuid[])        AS id,
 FROM (SELECT UNNEST(@ids::uuid[])AS id,
             UNNEST(@sent_ats::timestamptz[]) AS sent_at)
         AS new_values
 WHERE notification_messages.id = new_values.id;
diff --git a/coderd/notifications/enqueuer.go b/coderd/notifications/enqueuer.go
 }

 id := uuid.New()
 msg,err:= s.store.EnqueueNotificationMessage(ctx, database.EnqueueNotificationMessageParams{
 err = s.store.EnqueueNotificationMessage(ctx, database.EnqueueNotificationMessageParams{
 ID:                     id,
 UserID:                 userID,
 NotificationTemplateID: templateID,
 return nil, xerrors.Errorf("enqueue notification: %w", err)
 }

 s.log.Debug(ctx, "enqueued notification", slog.F("msg_id",msg.ID))
 s.log.Debug(ctx, "enqueued notification", slog.F("msg_id",id))
 return &id, nil
 }
Original file line number	Diff line number	Diff line change
Expand Up		@@ -983,6 +983,7 @@ func (r RootCmd) Server(newAPI func(context.Context, coderd.Options) (*coderd.
		)
		if experiments.Enabled(codersdk.ExperimentNotifications) {
		cfg := options.DeploymentValues.Notifications
		metrics := notifications.NewMetrics(options.PrometheusRegistry)

		// The enqueuer is responsible for enqueueing notifications to the given store.
		enqueuer, err := notifications.NewStoreEnqueuer(cfg, options.Database, templateHelpers(options), logger.Named("notifications.enqueuer"))
Expand All		@@ -994,7 +995,7 @@ func (r RootCmd) Server(newAPI func(context.Context, coderd.Options) (*coderd.
		// The notification manager is responsible for:
		// - creating notifiers and managing their lifecycles (notifiers are responsible for dequeueing/sending notifications)
		// - keeping the store updated with status updates
		notificationsManager, err = notifications.NewManager(cfg, options.Database, logger.Named("notifications.manager"))
		notificationsManager, err = notifications.NewManager(cfg, options.Database,metrics,logger.Named("notifications.manager"))
		if err != nil {
		return xerrors.Errorf("failed to instantiate notification manager: %w", err)
		}
Expand Down
Original file line number	Diff line number	Diff line change
Expand Up		@@ -1143,9 +1143,9 @@ func (q *querier) DeleteWorkspaceAgentPortSharesByTemplate(ctx context.Context,
		return q.db.DeleteWorkspaceAgentPortSharesByTemplate(ctx, templateID)
		}

		func (q *querier) EnqueueNotificationMessage(ctx context.Context, arg database.EnqueueNotificationMessageParams)(database.NotificationMessage,error) {
		func (q *querier) EnqueueNotificationMessage(ctx context.Context, arg database.EnqueueNotificationMessageParams) error {
		if err := q.authorizeContext(ctx, policy.ActionCreate, rbac.ResourceSystem); err != nil {
		returndatabase.NotificationMessage{},err
		return err
		}
		return q.db.EnqueueNotificationMessage(ctx, arg)
		}
Expand Down
Original file line number	Diff line number	Diff line change
Expand Up		@@ -935,12 +935,17 @@ func (q *FakeQuerier) AcquireNotificationMessages(_ context.Context, arg databas
		q.mutex.Lock()
		defer q.mutex.Unlock()

		var out []database.AcquireNotificationMessagesRow
		for _, nm:=rangeq.notificationMessages {
		iflen(out) >= int(arg.Count) {
		break
		}
		// Shift the first "Count" notifications off the slice (FIFO).
		sz:=len(q.notificationMessages)
dannykopping marked this conversation as resolved. Show resolvedHide resolved
		ifsz > int(arg.Count) {
		sz = int(arg.Count)
		}

		list := q.notificationMessages[:sz]
		q.notificationMessages = q.notificationMessages[sz:]

		var out []database.AcquireNotificationMessagesRow
		for _, nm := range list {
		acquirableStatuses := []database.NotificationMessageStatus{database.NotificationMessageStatusPending, database.NotificationMessageStatusTemporaryFailure}
		if !slices.Contains(acquirableStatuses, nm.Status) {
		continue
Expand All		@@ -956,9 +961,9 @@ func (q *FakeQuerier) AcquireNotificationMessages(_ context.Context, arg databas
		ID: nm.ID,
		Payload: nm.Payload,
		Method: nm.Method,
		CreatedBy: nm.CreatedBy,
		TitleTemplate: "This is a title with {{.Labels.variable}}",
		BodyTemplate: "This is a body with {{.Labels.variable}}",
		TemplateID: nm.NotificationTemplateID,
		})
		}

Expand DownExpand Up		@@ -1815,10 +1820,10 @@ func (q *FakeQuerier) DeleteWorkspaceAgentPortSharesByTemplate(_ context.Context
		return nil
		}

		func (q *FakeQuerier) EnqueueNotificationMessage(_ context.Context, arg database.EnqueueNotificationMessageParams)(database.NotificationMessage,error) {
		func (q *FakeQuerier) EnqueueNotificationMessage(_ context.Context, arg database.EnqueueNotificationMessageParams) error {
		err := validateDatabaseType(arg)
		if err != nil {
		returndatabase.NotificationMessage{},err
		return err
		}

		q.mutex.Lock()
Expand All		@@ -1827,7 +1832,7 @@ func (q *FakeQuerier) EnqueueNotificationMessage(_ context.Context, arg database
		var payload types.MessagePayload
		err = json.Unmarshal(arg.Payload, &payload)
		if err != nil {
		returndatabase.NotificationMessage{},err
		return err
		}

		nm := database.NotificationMessage{
Expand All		@@ -1845,7 +1850,7 @@ func (q *FakeQuerier) EnqueueNotificationMessage(_ context.Context, arg database

		q.notificationMessages = append(q.notificationMessages, nm)

		returnnm,err
		return err
		}

		func (q *FakeQuerier) FavoriteWorkspace(_ context.Context, arg uuid.UUID) error {
Expand Down
Original file line number	Diff line number	Diff line change
		@@ -0,0 +1,2 @@
		ALTER TABLE notification_messages
		DROP COLUMN IF EXISTS queued_seconds;
Original file line number	Diff line number	Diff line change
		@@ -0,0 +1,2 @@
		ALTER TABLE notification_messages
		ADD COLUMN queued_seconds FLOAT NULL;
Original file line number	Diff line number	Diff line change
Expand Up		@@ -10,16 +10,15 @@ FROM notification_templates nt,
		WHERE nt.id = @notification_template_id
		AND u.id = @user_id;

		-- name: EnqueueNotificationMessage :one
		-- name: EnqueueNotificationMessage :exec
		INSERT INTO notification_messages (id, notification_template_id, user_id, method, payload, targets, created_by)
		VALUES (@id,
		@notification_template_id,
		@user_id,
		@method::notification_method,
		@payload::jsonb,
		@targets,
		@created_by)
		RETURNING *;
		@created_by);

		-- Acquires the lease for a given count of notification messages, to enable concurrent dequeuing and subsequent sending.
		-- Only rows that aren't already leased (or ones which are leased but have exceeded their lease period) are returned.
Expand All		@@ -36,7 +35,8 @@ RETURNING *;
		WITH acquired AS (
		UPDATE
		notification_messages
		SET updated_at = NOW(),
		SET queued_seconds = GREATEST(0, EXTRACT(EPOCH FROM (NOW() - updated_at)))::FLOAT,
		updated_at = NOW(),
		status = 'leased'::notification_message_status,
		status_reason = 'Leased by notifier ' \|\| sqlc.arg('notifier_id')::uuid,
		leased_until = NOW() + CONCAT(sqlc.arg('lease_seconds')::int, ' seconds')::interval
Expand DownExpand Up		@@ -78,16 +78,19 @@ SELECT
		nm.id,
		nm.payload,
		nm.method,
		nm.created_by,
		nm.attempt_count::int AS attempt_count,
		nm.queued_seconds::float AS queued_seconds,
		-- template
		nt.id AS template_id,
		nt.title_template,
		nt.body_template
		FROM acquired nm
		JOIN notification_templates nt ON nm.notification_template_id = nt.id;

		-- name: BulkMarkNotificationMessagesFailed :execrows
		UPDATE notification_messages
		SET updated_at = subquery.failed_at,
		SET queued_seconds = 0,
		updated_at = subquery.failed_at,
		attempt_count = attempt_count + 1,
		status = CASE
		WHEN attempt_count + 1 < @max_attempts::int THEN subquery.status
Expand All		@@ -105,13 +108,14 @@ WHERE notification_messages.id = subquery.id;

		-- name: BulkMarkNotificationMessagesSent :execrows
		UPDATE notification_messages
		SET updated_at = new_values.sent_at,
		SET queued_seconds = 0,
		updated_at = new_values.sent_at,
		attempt_count = attempt_count + 1,
		status = 'sent'::notification_message_status,
		status_reason = NULL,
		leased_until = NULL,
		next_retry_after = NULL
		FROM (SELECT UNNEST(@ids::uuid[]) AS id,
		FROM (SELECT UNNEST(@ids::uuid[])AS id,
		UNNEST(@sent_ats::timestamptz[]) AS sent_at)
		AS new_values
		WHERE notification_messages.id = new_values.id;
Expand Down
Original file line number	Diff line number	Diff line change
Expand Up		@@ -59,7 +59,7 @@ func (s *StoreEnqueuer) Enqueue(ctx context.Context, userID, templateID uuid.UUI
		}

		id := uuid.New()
		msg,err:= s.store.EnqueueNotificationMessage(ctx, database.EnqueueNotificationMessageParams{
		err = s.store.EnqueueNotificationMessage(ctx, database.EnqueueNotificationMessageParams{
		ID: id,
		UserID: userID,
		NotificationTemplateID: templateID,
Expand All		@@ -73,7 +73,7 @@ func (s *StoreEnqueuer) Enqueue(ctx context.Context, userID, templateID uuid.UUI
		return nil, xerrors.Errorf("enqueue notification: %w", err)
		}

		s.log.Debug(ctx, "enqueued notification", slog.F("msg_id",msg.ID))
		s.log.Debug(ctx, "enqueued notification", slog.F("msg_id",id))
		return &id, nil
		}

Expand Down