loop contains the main business logic of the notification manager. It is responsible for subscribing to notification events, creating a notifier, and publishing bulk dispatch result updates to the store.
(ctx context.Context)
| 161 | // loop contains the main business logic of the notification manager. It is responsible for subscribing to notification |
| 162 | // events, creating a notifier, and publishing bulk dispatch result updates to the store. |
| 163 | func (m *Manager) loop(ctx context.Context) error { |
| 164 | defer func() { |
| 165 | close(m.done) |
| 166 | m.log.Debug(context.Background(), "notification manager stopped") |
| 167 | }() |
| 168 | |
| 169 | m.mu.Lock() |
| 170 | if m.closed { |
| 171 | m.mu.Unlock() |
| 172 | return ErrManagerAlreadyClosed |
| 173 | } |
| 174 | |
| 175 | var eg errgroup.Group |
| 176 | |
| 177 | m.notifier = newNotifier(ctx, m.cfg, uuid.New(), m.log, m.store, m.handlers, m.helpers, m.metrics, m.clock) |
| 178 | eg.Go(func() error { |
| 179 | // run the notifier which will handle dequeueing and dispatching notifications. |
| 180 | return m.notifier.run(m.success, m.failure) |
| 181 | }) |
| 182 | |
| 183 | m.mu.Unlock() |
| 184 | |
| 185 | // Periodically flush notification state changes to the store. |
| 186 | eg.Go(func() error { |
| 187 | // Every interval, collect the messages in the channels and bulk update them in the store. |
| 188 | tick := m.clock.NewTicker(m.cfg.StoreSyncInterval.Value(), "Manager", "storeSync") |
| 189 | defer tick.Stop() |
| 190 | for { |
| 191 | select { |
| 192 | case <-ctx.Done(): |
| 193 | // Nothing we can do in this scenario except bail out; after the message lease expires, the messages will |
| 194 | // be requeued and users will receive duplicates. |
| 195 | // This is an explicit trade-off between keeping the database load light (by bulk-updating records) and |
| 196 | // exactly-once delivery. |
| 197 | // |
| 198 | // The current assumption is that duplicate delivery of these messages is, at worst, slightly annoying. |
| 199 | // If these notifications are triggering external actions (e.g. via webhooks) this could be more |
| 200 | // consequential, and we may need a more sophisticated mechanism. |
| 201 | // |
| 202 | // TODO: mention the above tradeoff in documentation. |
| 203 | m.log.Warn(ctx, "exiting ungracefully", slog.Error(ctx.Err())) |
| 204 | |
| 205 | if len(m.success)+len(m.failure) > 0 { |
| 206 | m.log.Warn(ctx, "content canceled with pending updates in buffer, these messages will be sent again after lease expires", |
| 207 | slog.F("success_count", len(m.success)), slog.F("failure_count", len(m.failure))) |
| 208 | } |
| 209 | return ctx.Err() |
| 210 | case <-m.stop: |
| 211 | if len(m.success)+len(m.failure) > 0 { |
| 212 | m.log.Warn(ctx, "flushing buffered updates before stop", |
| 213 | slog.F("success_count", len(m.success)), slog.F("failure_count", len(m.failure))) |
| 214 | m.syncUpdates(ctx) |
| 215 | m.log.Warn(ctx, "flushing updates done") |
| 216 | } |
| 217 | return nil |
| 218 | case <-tick.C: |
| 219 | m.syncUpdates(ctx) |
| 220 | } |