watchNotificationsSMTP polls the SMTP HTTP API for notifications and returns error or nil once all expected notifications are received.
(ctx context.Context, user codersdk.User, logger slog.Logger, expectedNotifications map[uuid.UUID]struct{})
| 290 | // watchNotificationsSMTP polls the SMTP HTTP API for notifications and returns error or nil |
| 291 | // once all expected notifications are received. |
| 292 | func (r *Runner) watchNotificationsSMTP(ctx context.Context, user codersdk.User, logger slog.Logger, expectedNotifications map[uuid.UUID]struct{}) error { |
| 293 | logger.Info(ctx, "polling SMTP API for notifications", |
| 294 | slog.F("email", user.Email), |
| 295 | slog.F("expected_count", len(expectedNotifications)), |
| 296 | ) |
| 297 | receivedNotifications := make(map[uuid.UUID]struct{}) |
| 298 | |
| 299 | apiURL := fmt.Sprintf("%s/messages?email=%s", r.cfg.SMTPApiURL, user.Email) |
| 300 | httpClient := r.cfg.SMTPHttpClient |
| 301 | |
| 302 | const smtpPollInterval = 2 * time.Second |
| 303 | done := xerrors.New("done") |
| 304 | |
| 305 | tkr := r.clock.TickerFunc(ctx, smtpPollInterval, func() error { |
| 306 | reqCtx, cancel := context.WithTimeout(ctx, r.cfg.SMTPRequestTimeout) |
| 307 | defer cancel() |
| 308 | |
| 309 | req, err := http.NewRequestWithContext(reqCtx, http.MethodGet, apiURL, nil) |
| 310 | if err != nil { |
| 311 | logger.Error(ctx, "create SMTP API request", slog.Error(err)) |
| 312 | r.cfg.Metrics.AddError("smtp_create_request") |
| 313 | return xerrors.Errorf("create SMTP API request: %w", err) |
| 314 | } |
| 315 | |
| 316 | resp, err := httpClient.Do(req) |
| 317 | if err != nil { |
| 318 | logger.Error(ctx, "poll smtp api for notifications", slog.Error(err)) |
| 319 | r.cfg.Metrics.AddError("smtp_poll") |
| 320 | return nil |
| 321 | } |
| 322 | |
| 323 | if resp.StatusCode != http.StatusOK { |
| 324 | // discard the response to allow reusing of the connection |
| 325 | _, _ = io.Copy(io.Discard, resp.Body) |
| 326 | _ = resp.Body.Close() |
| 327 | logger.Error(ctx, "smtp api returned non-200 status", slog.F("status", resp.StatusCode)) |
| 328 | r.cfg.Metrics.AddError("smtp_bad_status") |
| 329 | return nil |
| 330 | } |
| 331 | |
| 332 | var summaries []smtpmock.EmailSummary |
| 333 | if err := json.NewDecoder(resp.Body).Decode(&summaries); err != nil { |
| 334 | _ = resp.Body.Close() |
| 335 | logger.Error(ctx, "decode smtp api response", slog.Error(err)) |
| 336 | r.cfg.Metrics.AddError("smtp_decode") |
| 337 | return xerrors.Errorf("decode smtp api response: %w", err) |
| 338 | } |
| 339 | _ = resp.Body.Close() |
| 340 | |
| 341 | // Process each email summary |
| 342 | for _, summary := range summaries { |
| 343 | notificationID := summary.NotificationTemplateID |
| 344 | if notificationID == uuid.Nil { |
| 345 | continue |
| 346 | } |
| 347 | |
| 348 | if _, exists := expectedNotifications[notificationID]; exists { |
| 349 | if _, received := receivedNotifications[notificationID]; !received { |