diff --git a/pkg/coredata/email.go b/pkg/coredata/email.go index 822d1cf04..075967fdd 100644 --- a/pkg/coredata/email.go +++ b/pkg/coredata/email.go @@ -28,15 +28,21 @@ import ( type ( Email struct { - ID gid.GID `db:"id"` - RecipientEmail string `db:"recipient_email"` - RecipientName string `db:"recipient_name"` - Subject string `db:"subject"` - TextBody string `db:"text_body"` - HtmlBody *string `db:"html_body"` - CreatedAt time.Time `db:"created_at"` - UpdatedAt time.Time `db:"updated_at"` - SentAt *time.Time `db:"sent_at"` + ID gid.GID `db:"id"` + RecipientEmail string `db:"recipient_email"` + RecipientName string `db:"recipient_name"` + Subject string `db:"subject"` + TextBody string `db:"text_body"` + HtmlBody *string `db:"html_body"` + Status EmailStatus `db:"status"` + ProcessingStartedAt *time.Time `db:"processing_started_at"` + AttemptCount int `db:"attempt_count"` + MaxAttempts int `db:"max_attempts"` + LastAttemptedAt *time.Time `db:"last_attempted_at"` + LastError *string `db:"last_error"` + CreatedAt time.Time `db:"created_at"` + UpdatedAt time.Time `db:"updated_at"` + SentAt *time.Time `db:"sent_at"` } ) @@ -65,6 +71,9 @@ func NewEmail( Subject: subject, TextBody: textBody, HtmlBody: htmlBody, + Status: EmailStatusPending, + AttemptCount: 0, + MaxAttempts: 10, CreatedAt: now, UpdatedAt: now, } @@ -75,8 +84,14 @@ func (e *Email) Insert( conn pg.Conn, ) error { q := ` -INSERT INTO emails (id, recipient_email, recipient_name, subject, text_body, html_body, created_at, updated_at) -VALUES (@id, @recipient_email, @recipient_name, @subject, @text_body, @html_body, @created_at, @updated_at) +INSERT INTO emails ( + id, recipient_email, recipient_name, subject, text_body, html_body, + status, attempt_count, max_attempts, created_at, updated_at +) +VALUES ( + @id, @recipient_email, @recipient_name, @subject, @text_body, @html_body, + @status, @attempt_count, @max_attempts, @created_at, @updated_at +) ` args := pgx.StrictNamedArgs{ @@ -86,6 +101,9 @@ VALUES (@id, @recipient_email, @recipient_name, @subject, @text_body, @html_body "subject": e.Subject, "text_body": e.TextBody, "html_body": e.HtmlBody, + "status": e.Status, + "attempt_count": e.AttemptCount, + "max_attempts": e.MaxAttempts, "created_at": e.CreatedAt, "updated_at": e.UpdatedAt, } @@ -94,17 +112,20 @@ VALUES (@id, @recipient_email, @recipient_name, @subject, @text_body, @html_body return err } -func (e *Email) LoadNextUnsentForUpdate( +func (e *Email) LoadNextPendingForUpdateSkipLocked( ctx context.Context, conn pg.Conn, ) error { q := ` -SELECT id, recipient_email, recipient_name, subject, text_body, html_body, created_at, updated_at, sent_at +SELECT + id, recipient_email, recipient_name, subject, text_body, html_body, + status, processing_started_at, attempt_count, max_attempts, + last_attempted_at, last_error, created_at, updated_at, sent_at FROM emails -WHERE sent_at IS NULL +WHERE status = 'PENDING' AND attempt_count < max_attempts ORDER BY created_at ASC LIMIT 1 -FOR UPDATE +FOR UPDATE SKIP LOCKED ` rows, err := conn.Query(ctx, q) @@ -132,16 +153,47 @@ func (e *Email) Update( ) error { q := ` UPDATE emails -SET sent_at = @sent_at, updated_at = @updated_at +SET + status = @status, + processing_started_at = @processing_started_at, + attempt_count = @attempt_count, + last_attempted_at = @last_attempted_at, + last_error = @last_error, + sent_at = @sent_at, + updated_at = @updated_at WHERE id = @id ` args := pgx.StrictNamedArgs{ - "id": e.ID, - "sent_at": e.SentAt, - "updated_at": e.UpdatedAt, + "id": e.ID, + "status": e.Status, + "processing_started_at": e.ProcessingStartedAt, + "attempt_count": e.AttemptCount, + "last_attempted_at": e.LastAttemptedAt, + "last_error": e.LastError, + "sent_at": e.SentAt, + "updated_at": e.UpdatedAt, } _, err := conn.Exec(ctx, q, args) return err } + +func ResetStaleProcessingEmails( + ctx context.Context, + conn pg.Conn, + staleAfter time.Duration, +) error { + q := ` +UPDATE emails +SET status = 'PENDING', processing_started_at = NULL, updated_at = NOW() +WHERE status = 'PROCESSING' + AND processing_started_at < NOW() - $1::interval +` + _, err := conn.Exec(ctx, q, staleAfter) + if err != nil { + return fmt.Errorf("cannot reset stale processing emails: %w", err) + } + + return nil +} diff --git a/pkg/coredata/email_status.go b/pkg/coredata/email_status.go new file mode 100644 index 000000000..8dc1760ac --- /dev/null +++ b/pkg/coredata/email_status.go @@ -0,0 +1,71 @@ +// Copyright (c) 2025 Probo Inc . +// +// Permission to use, copy, modify, and/or distribute this software for any +// purpose with or without fee is hereby granted, provided that the above +// copyright notice and this permission notice appear in all copies. +// +// THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH +// REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY +// AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, +// INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM +// LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR +// OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR +// PERFORMANCE OF THIS SOFTWARE. + +package coredata + +import ( + "database/sql/driver" + "fmt" +) + +type ( + EmailStatus string +) + +const ( + EmailStatusPending EmailStatus = "PENDING" + EmailStatusProcessing EmailStatus = "PROCESSING" + EmailStatusSent EmailStatus = "SENT" + EmailStatusFailed EmailStatus = "FAILED" +) + +func (s EmailStatus) MarshalText() ([]byte, error) { + return []byte(s.String()), nil +} + +func (s *EmailStatus) UnmarshalText(data []byte) error { + val := string(data) + + switch val { + case EmailStatusPending.String(): + *s = EmailStatusPending + case EmailStatusProcessing.String(): + *s = EmailStatusProcessing + case EmailStatusSent.String(): + *s = EmailStatusSent + case EmailStatusFailed.String(): + *s = EmailStatusFailed + default: + return fmt.Errorf("invalid EmailStatus value: %q", val) + } + + return nil +} + +func (s EmailStatus) String() string { + return string(s) +} + +func (s *EmailStatus) Scan(value any) error { + val, ok := value.(string) + if !ok { + return fmt.Errorf("invalid scan source for EmailStatus, expected string got %T", value) + } + + return s.UnmarshalText([]byte(val)) +} + +func (s EmailStatus) Value() (driver.Value, error) { + return s.String(), nil +} diff --git a/pkg/coredata/migrations/20260308T120000Z.sql b/pkg/coredata/migrations/20260308T120000Z.sql new file mode 100644 index 000000000..46695bc01 --- /dev/null +++ b/pkg/coredata/migrations/20260308T120000Z.sql @@ -0,0 +1,12 @@ +ALTER TABLE emails +ADD COLUMN status TEXT NOT NULL DEFAULT 'PENDING', +ADD COLUMN processing_started_at TIMESTAMP WITH TIME ZONE, +ADD COLUMN attempt_count INTEGER NOT NULL DEFAULT 0, +ADD COLUMN max_attempts INTEGER NOT NULL DEFAULT 10, +ADD COLUMN last_attempted_at TIMESTAMP WITH TIME ZONE, +ADD COLUMN last_error TEXT; + +UPDATE emails SET status = 'SENT' WHERE sent_at IS NOT NULL; + +CREATE INDEX idx_emails_pending ON emails (status, created_at) +WHERE status = 'PENDING'; diff --git a/pkg/mailer/mailer.go b/pkg/mailer/mailer.go index bf7bf5bf2..be9fb7c2a 100644 --- a/pkg/mailer/mailer.go +++ b/pkg/mailer/mailer.go @@ -22,6 +22,7 @@ import ( "fmt" "net" "net/smtp" + "sync" "time" "github.com/jhillyerd/enmime" @@ -32,90 +33,333 @@ import ( ) type ( - Mailer struct { - pg *pg.Client - fileManager *filemanager.Service - l *log.Logger - cfg Config - interval time.Duration + SendingWorker struct { + pg *pg.Client + fileManager *filemanager.Service + logger *log.Logger + smtp SMTPConfig + senderName string + senderEmail string + interval time.Duration + smtpTimeout time.Duration + staleAfter time.Duration + maxConcurrency int } - Config struct { - SenderName string - SenderEmail string + SMTPConfig struct { Addr string - Timeout time.Duration User string Password string TLSRequired bool - Interval time.Duration } + + SendingWorkerOption func(*SendingWorker) ) -func NewMailer(pg *pg.Client, fileManager *filemanager.Service, l *log.Logger, cfg Config) *Mailer { - // Set a default timeout if not provided - if cfg.Timeout == 0 { - cfg.Timeout = 10 * time.Second - } - - // Set a default interval if not provided - if cfg.Interval == 0 { - cfg.Interval = 60 * time.Second - } - - return &Mailer{pg: pg, fileManager: fileManager, l: l, cfg: cfg, interval: cfg.Interval} +func WithSendingWorkerInterval(d time.Duration) SendingWorkerOption { + return func(w *SendingWorker) { w.interval = d } } -func (m *Mailer) Run(ctx context.Context) error { -LOOP: +func WithSendingWorkerSMTPTimeout(d time.Duration) SendingWorkerOption { + return func(w *SendingWorker) { w.smtpTimeout = d } +} + +func WithSendingWorkerStaleAfter(d time.Duration) SendingWorkerOption { + return func(w *SendingWorker) { w.staleAfter = d } +} + +func WithSendingWorkerMaxConcurrency(n int) SendingWorkerOption { + return func(w *SendingWorker) { + if n > 0 { + w.maxConcurrency = n + } + } +} + +func NewSendingWorker( + pgClient *pg.Client, + fileManager *filemanager.Service, + senderName string, + senderEmail string, + smtpCfg SMTPConfig, + logger *log.Logger, + opts ...SendingWorkerOption, +) *SendingWorker { + w := &SendingWorker{ + pg: pgClient, + fileManager: fileManager, + logger: logger, + smtp: smtpCfg, + senderName: senderName, + senderEmail: senderEmail, + interval: 30 * time.Second, + smtpTimeout: 25 * time.Second, + staleAfter: 5 * time.Minute, + maxConcurrency: 20, + } + + for _, opt := range opts { + opt(w) + } + + return w +} + +func (w *SendingWorker) Run(ctx context.Context) error { + var ( + wg sync.WaitGroup + sem = make(chan struct{}, w.maxConcurrency) + ) + + defer wg.Wait() + + ticker := time.NewTicker(w.interval) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + return ctx.Err() + case <-ticker.C: + nonCancelableCtx := context.WithoutCancel(ctx) + w.recoverStaleRows(nonCancelableCtx) + + for { + if err := w.processNext(ctx, sem, &wg); err != nil { + if !errors.Is(err, coredata.ErrNoUnsentEmail) { + w.logger.ErrorCtx(nonCancelableCtx, "cannot process email", log.Error(err)) + } + break + } + } + } + } +} + +func (w *SendingWorker) processNext(ctx context.Context, sem chan struct{}, wg *sync.WaitGroup) error { select { + case sem <- struct{}{}: case <-ctx.Done(): return ctx.Err() - case <-time.After(m.interval): - ctx := context.Background() - if err := m.batchSendEmails(ctx); err != nil { - m.l.ErrorCtx(ctx, "cannot send email", log.Error(err)) - } + } - goto LOOP + var ( + email = coredata.Email{} + nonCancelableCtx = context.WithoutCancel(ctx) + ) + + if err := w.pg.WithTx( + nonCancelableCtx, + func(tx pg.Conn) error { + if err := email.LoadNextPendingForUpdateSkipLocked(nonCancelableCtx, tx); err != nil { + return err + } + + now := time.Now() + email.Status = coredata.EmailStatusProcessing + email.ProcessingStartedAt = &now + email.AttemptCount++ + email.LastAttemptedAt = &now + email.UpdatedAt = now + + if err := email.Update(nonCancelableCtx, tx); err != nil { + return fmt.Errorf("cannot update email: %w", err) + } + + return nil + }, + ); err != nil { + <-sem + return err + } + + wg.Add(1) + go func(email coredata.Email) { + defer wg.Done() + defer func() { <-sem }() + + if sendErr := w.sendAndCommit(nonCancelableCtx, &email); sendErr != nil { + if failErr := w.failEmail(nonCancelableCtx, &email, sendErr); failErr != nil { + w.logger.ErrorCtx(nonCancelableCtx, "cannot fail email", log.Error(failErr)) + } + } + }(email) + + return nil +} + +func (w *SendingWorker) sendAndCommit( + ctx context.Context, + email *coredata.Email, +) error { + var buf bytes.Buffer + + if err := w.pg.WithConn( + ctx, + func(conn pg.Conn) error { + var attachments coredata.EmailAttachments + if err := attachments.LoadByEmailID(ctx, conn, email.ID); err != nil { + return fmt.Errorf("cannot load email attachments: %w", err) + } + + mail := enmime.Builder(). + Subject(email.Subject). + From(w.senderName, w.senderEmail). + To(email.RecipientName, email.RecipientEmail). + Text([]byte(email.TextBody)) + + if email.HtmlBody != nil { + mail = mail.HTML([]byte(*email.HtmlBody)) + } + + for _, att := range attachments { + var file coredata.File + if err := file.LoadByID(ctx, conn, coredata.NewNoScope(), att.FileID); err != nil { + return fmt.Errorf("cannot load file record for attachment %s: %w", att.Filename, err) + } + + data, err := w.fileManager.GetFileBytes(ctx, &file) + if err != nil { + return fmt.Errorf("cannot download attachment %s: %w", att.Filename, err) + } + + mail = mail.AddAttachment(data, file.MimeType, att.Filename) + } + + envelope, err := mail.Build() + if err != nil { + return fmt.Errorf("cannot build email: %w", err) + } + + if err := envelope.Encode(&buf); err != nil { + return fmt.Errorf("cannot encode email: %w", err) + } + + return nil + }, + ); err != nil { + return err + } + + sendCtx, cancel := context.WithTimeout(ctx, w.smtpTimeout) + defer cancel() + + if err := w.sendMail(sendCtx, []string{email.RecipientEmail}, buf.Bytes()); err != nil { + if errors.Is(err, context.DeadlineExceeded) { + return fmt.Errorf("email sending timed out after %s: %w", w.smtpTimeout, err) + } + return fmt.Errorf("cannot send email: %w", err) + } + + if err := w.pg.WithTx( + ctx, + func(tx pg.Conn) error { + now := time.Now() + email.Status = coredata.EmailStatusSent + email.SentAt = &now + email.ProcessingStartedAt = nil + email.LastError = nil + email.UpdatedAt = now + + if err := email.Update(ctx, tx); err != nil { + return fmt.Errorf("cannot update email: %w", err) + } + + return nil + }, + ); err != nil { + w.logger.ErrorCtx(ctx, + "email sent but failed to commit status update; will not re-queue to avoid duplicate delivery", + log.Error(err), + log.String("email_id", email.ID.String()), + ) + } + + return nil +} + +func (w *SendingWorker) failEmail( + ctx context.Context, + email *coredata.Email, + processingError error, +) error { + w.logger.ErrorCtx(ctx, "sending worker failure", + log.Error(processingError), + log.String("email_id", email.ID.String()), + ) + + return w.pg.WithTx( + ctx, + func(tx pg.Conn) error { + errStr := processingError.Error() + email.LastError = &errStr + email.ProcessingStartedAt = nil + email.UpdatedAt = time.Now() + + if email.AttemptCount >= email.MaxAttempts { + email.Status = coredata.EmailStatusFailed + } else { + email.Status = coredata.EmailStatusPending + } + + if err := email.Update(ctx, tx); err != nil { + return fmt.Errorf("cannot update email: %w", err) + } + + return nil + }, + ) +} + +func (w *SendingWorker) recoverStaleRows(ctx context.Context) { + if err := w.pg.WithConn( + ctx, + func(conn pg.Conn) error { + return coredata.ResetStaleProcessingEmails(ctx, conn, w.staleAfter) + }, + ); err != nil { + w.logger.ErrorCtx(ctx, "cannot recover stale emails", log.Error(err)) } } -func (m *Mailer) sendMailWithTimeout(ctx context.Context, to []string, msg []byte) error { - host, _, err := net.SplitHostPort(m.cfg.Addr) +func (w *SendingWorker) sendMail(ctx context.Context, to []string, msg []byte) error { + host, _, err := net.SplitHostPort(w.smtp.Addr) if err != nil { return fmt.Errorf("invalid address: %w", err) } var d net.Dialer - d.Timeout = 5 * time.Second - conn, err := d.DialContext(ctx, "tcp", m.cfg.Addr) + conn, err := d.DialContext(ctx, "tcp", w.smtp.Addr) if err != nil { return fmt.Errorf("connection error: %w", err) } - defer func() { _ = conn.Close() }() + + if deadline, ok := ctx.Deadline(); ok { + conn.SetDeadline(deadline) + } c, err := smtp.NewClient(conn, host) if err != nil { + _ = conn.Close() return fmt.Errorf("SMTP client creation error: %w", err) } defer func() { _ = c.Quit() }() - if m.cfg.TLSRequired { + if w.smtp.TLSRequired { if err := c.StartTLS(&tls.Config{ServerName: host}); err != nil { return fmt.Errorf("TLS negotiation error: %w", err) } } - if m.cfg.User != "" && m.cfg.Password != "" { - auth := smtp.PlainAuth("", m.cfg.User, m.cfg.Password, host) + if w.smtp.User != "" && w.smtp.Password != "" { + auth := smtp.PlainAuth("", w.smtp.User, w.smtp.Password, host) if err = c.Auth(auth); err != nil { return fmt.Errorf("SMTP authentication error: %w", err) } } - if err = c.Mail(m.cfg.SenderEmail); err != nil { + if err = c.Mail(w.senderEmail); err != nil { return fmt.Errorf("MAIL FROM error: %w", err) } @@ -125,103 +369,19 @@ func (m *Mailer) sendMailWithTimeout(ctx context.Context, to []string, msg []byt } } - w, err := c.Data() + wr, err := c.Data() if err != nil { return fmt.Errorf("DATA command error: %w", err) } - _, err = w.Write(msg) + _, err = wr.Write(msg) if err != nil { return fmt.Errorf("message write error: %w", err) } - if err = w.Close(); err != nil { + if err = wr.Close(); err != nil { return fmt.Errorf("message close error: %w", err) } - return c.Quit() -} - -func (m *Mailer) batchSendEmails(ctx context.Context) error { - for { - err := m.pg.WithTx( - ctx, - func(tx pg.Conn) error { - email := &coredata.Email{} - err := email.LoadNextUnsentForUpdate(ctx, tx) - if err != nil { - return err - } - - var attachments coredata.EmailAttachments - if err := attachments.LoadByEmailID(ctx, tx, email.ID); err != nil { - return fmt.Errorf("cannot load email attachments: %w", err) - } - - mail := enmime.Builder(). - Subject(email.Subject). - From(m.cfg.SenderName, m.cfg.SenderEmail). - To(email.RecipientName, email.RecipientEmail). - Text([]byte(email.TextBody)) - - if email.HtmlBody != nil { - mail = mail.HTML([]byte(*email.HtmlBody)) - } - - for _, att := range attachments { - var file coredata.File - if err := file.LoadByID(ctx, tx, coredata.NewNoScope(), att.FileID); err != nil { - return fmt.Errorf("cannot load file record for attachment %s: %w", att.Filename, err) - } - - // TODO use reader - data, err := m.fileManager.GetFileBytes(ctx, &file) - if err != nil { - return fmt.Errorf("cannot download attachment %s: %w", att.Filename, err) - } - - // TODO [esign] maybe use AddAttachmentWithReader instead? - mail = mail.AddAttachment(data, file.MimeType, att.Filename) - } - - envelope, err := mail.Build() - if err != nil { - return fmt.Errorf("cannot build email: %w", err) - } - - var buf bytes.Buffer - if err := envelope.Encode(&buf); err != nil { - return fmt.Errorf("cannot encode email: %w", err) - } - - sendCtx, cancel := context.WithTimeout(ctx, m.cfg.Timeout) - defer cancel() - - if err := m.sendMailWithTimeout(sendCtx, []string{email.RecipientEmail}, buf.Bytes()); err != nil { - if errors.Is(err, context.DeadlineExceeded) { - return fmt.Errorf("email sending timed out after %s: %w", m.cfg.Timeout, err) - } - return fmt.Errorf("cannot send email: %w", err) - } - - now := time.Now() - email.SentAt = &now - email.UpdatedAt = now - - if err := email.Update(ctx, tx); err != nil { - return fmt.Errorf("cannot update email: %w", err) - } - - return nil - }, - ) - - if errors.Is(err, coredata.ErrNoUnsentEmail) { - return nil - } - - if err != nil { - return err - } - } + return nil } diff --git a/pkg/probod/probod.go b/pkg/probod/probod.go index 51610bc1a..3de317763 100644 --- a/pkg/probod/probod.go +++ b/pkg/probod/probod.go @@ -514,25 +514,25 @@ func (impl *Implm) Run( ) mailerCtx, stopMailer := context.WithCancel(context.Background()) - mailer := mailer.NewMailer( + sendingWorker := mailer.NewSendingWorker( pgClient, fileManagerService, - l, - mailer.Config{ - SenderEmail: impl.cfg.Notifications.Mailer.SenderEmail, - SenderName: impl.cfg.Notifications.Mailer.SenderName, + impl.cfg.Notifications.Mailer.SenderName, + impl.cfg.Notifications.Mailer.SenderEmail, + mailer.SMTPConfig{ Addr: impl.cfg.Notifications.Mailer.SMTP.Addr, User: impl.cfg.Notifications.Mailer.SMTP.User, Password: impl.cfg.Notifications.Mailer.SMTP.Password, TLSRequired: impl.cfg.Notifications.Mailer.SMTP.TLSRequired, - Timeout: time.Second * 10, - Interval: time.Duration(impl.cfg.Notifications.Mailer.MailerInterval) * time.Second, }, + l.Named("sending-worker"), + mailer.WithSendingWorkerSMTPTimeout(time.Second*10), + mailer.WithSendingWorkerInterval(time.Duration(impl.cfg.Notifications.Mailer.MailerInterval)*time.Second), ) wg.Go( func() { - if err := mailer.Run(mailerCtx); err != nil { - cancel(fmt.Errorf("mailer crashed: %w", err)) + if err := sendingWorker.Run(mailerCtx); err != nil { + cancel(fmt.Errorf("sending worker crashed: %w", err)) } }, )