watchNotifications reads notifications from the websocket and returns error or nil once all expected notifications are received.
(ctx context.Context, conn *websocket.Conn, user codersdk.User, logger slog.Logger, expectedNotifications map[uuid.UUID]struct{})
| 240 | // watchNotifications reads notifications from the websocket and returns error or nil |
| 241 | // once all expected notifications are received. |
| 242 | func (r *Runner) watchNotifications(ctx context.Context, conn *websocket.Conn, user codersdk.User, logger slog.Logger, expectedNotifications map[uuid.UUID]struct{}) error { |
| 243 | logger.Info(ctx, "waiting for notifications", |
| 244 | slog.F("username", user.Username), |
| 245 | slog.F("expected_count", len(expectedNotifications))) |
| 246 | |
| 247 | receivedNotifications := make(map[uuid.UUID]struct{}) |
| 248 | |
| 249 | for { |
| 250 | select { |
| 251 | case <-ctx.Done(): |
| 252 | return xerrors.Errorf("context canceled while waiting for notifications: %w", ctx.Err()) |
| 253 | default: |
| 254 | } |
| 255 | |
| 256 | if len(receivedNotifications) == len(expectedNotifications) { |
| 257 | logger.Info(ctx, "received all expected notifications") |
| 258 | return nil |
| 259 | } |
| 260 | |
| 261 | notif, err := readNotification(ctx, conn) |
| 262 | if err != nil { |
| 263 | logger.Error(ctx, "read notification", slog.Error(err)) |
| 264 | r.cfg.Metrics.AddError("read_notification_websocket") |
| 265 | return xerrors.Errorf("read notification: %w", err) |
| 266 | } |
| 267 | |
| 268 | templateID := notif.Notification.TemplateID |
| 269 | if _, exists := expectedNotifications[templateID]; exists { |
| 270 | if _, received := receivedNotifications[templateID]; !received { |
| 271 | receiptTime := time.Now() |
| 272 | r.websocketReceiptTimesMu.Lock() |
| 273 | r.websocketReceiptTimes[templateID] = receiptTime |
| 274 | r.websocketReceiptTimesMu.Unlock() |
| 275 | receivedNotifications[templateID] = struct{}{} |
| 276 | |
| 277 | logger.Info(ctx, "received expected notification", |
| 278 | slog.F("template_id", templateID), |
| 279 | slog.F("title", notif.Notification.Title), |
| 280 | slog.F("receipt_time", receiptTime)) |
| 281 | } |
| 282 | } else { |
| 283 | logger.Debug(ctx, "received notification not being tested", |
| 284 | slog.F("template_id", templateID), |
| 285 | slog.F("title", notif.Notification.Title)) |
| 286 | } |
| 287 | } |
| 288 | } |
| 289 | |
| 290 | // watchNotificationsSMTP polls the SMTP HTTP API for notifications and returns error or nil |
| 291 | // once all expected notifications are received. |