Method syncUpdates

coderd/notifications/manager.go:238–356

syncUpdates updates messages in the store based on the given successful and failed message dispatch results.

(ctx context.Context)

// syncUpdates updates messages in the store based on the given successful and failed message dispatch results.
func (m *Manager) syncUpdates(ctx context.Context) {
	// Ensure we update the metrics to reflect the current state after each invocation.
	defer func() {
		m.metrics.PendingUpdates.Set(float64(len(m.success) + len(m.failure)))
	}()

	select {
	case <-ctx.Done():
		return
	default:
	}

	nSuccess := len(m.success)
	nFailure := len(m.failure)

	m.metrics.PendingUpdates.Set(float64(nSuccess + nFailure))

	// Nothing to do.
	if nSuccess+nFailure == 0 {
		return
	}

	var (
		successParams database.BulkMarkNotificationMessagesSentParams
		failureParams database.BulkMarkNotificationMessagesFailedParams
	)

	// Read all the existing messages due for update from the channel, but don't range over the channels because they
	// block until they are closed.
	//
	// This is vulnerable to TOCTOU, but it's fine.
	// If more items are added to the success or failure channels between measuring their lengths and now, those items
	// will be processed on the next bulk update.

	for i := 0; i < nSuccess; i++ {
		res := <-m.success
		successParams.IDs = append(successParams.IDs, res.msg)
		successParams.SentAts = append(successParams.SentAts, res.ts)
	}
	for i := 0; i < nFailure; i++ {
		res := <-m.failure

		var (
			reason string
			status database.NotificationMessageStatus
		)

		switch {
		case res.retryable:
			status = database.NotificationMessageStatusTemporaryFailure
		case res.inhibited:
			status = database.NotificationMessageStatusInhibited
			reason = "disabled by user"
		default:
			status = database.NotificationMessageStatusPermanentFailure
		}

		failureParams.IDs = append(failureParams.IDs, res.msg)