Improve mailer performance

Signed-off-by: Bryan Frimin <bryan@getprobo.com>

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
+}

pkg/coredata/email_status.go

@@ -0,0 +1,71 @@
+// Copyright (c) 2025 Probo Inc <hello@getprobo.com>.
+//
+// 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
+}

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';