345 lines
9.8 KiB
Go
345 lines
9.8 KiB
Go
package email
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"fmt"
|
|
"log/slog"
|
|
"math"
|
|
"time"
|
|
)
|
|
|
|
// Queue stores outgoing emails durably and dispatches them via a Sender.
|
|
// It is safe for concurrent use; claim uses an atomic UPDATE ... WHERE
|
|
// status='pending' to lease rows without races.
|
|
type Queue struct {
|
|
db *sql.DB
|
|
sender Sender
|
|
logger *slog.Logger
|
|
|
|
pollInterval time.Duration
|
|
maxBackoff time.Duration
|
|
}
|
|
|
|
// QueueRow is the on-disk representation of a queued email.
|
|
type QueueRow struct {
|
|
ID string
|
|
NotificationID string
|
|
ToAddress string
|
|
Subject string
|
|
Body string
|
|
Status string
|
|
Attempts int
|
|
MaxAttempts int
|
|
LastError string
|
|
NextAttemptAt time.Time
|
|
SentAt *time.Time
|
|
CreatedAt time.Time
|
|
}
|
|
|
|
// NewQueue constructs a queue. sender may be nil; in that case emails stay
|
|
// pending forever (useful for tests that only inspect enqueue state).
|
|
func NewQueue(db *sql.DB, sender Sender, logger *slog.Logger) *Queue {
|
|
if logger == nil {
|
|
logger = slog.Default()
|
|
}
|
|
return &Queue{
|
|
db: db,
|
|
sender: sender,
|
|
logger: logger,
|
|
pollInterval: 10 * time.Second,
|
|
maxBackoff: 30 * time.Minute,
|
|
}
|
|
}
|
|
|
|
// SetPollInterval overrides the worker poll interval. Must be called before
|
|
// Run. Useful for tests.
|
|
func (q *Queue) SetPollInterval(d time.Duration) {
|
|
if d > 0 {
|
|
q.pollInterval = d
|
|
}
|
|
}
|
|
|
|
// SetMaxBackoff overrides the maximum retry backoff.
|
|
func (q *Queue) SetMaxBackoff(d time.Duration) {
|
|
if d > 0 {
|
|
q.maxBackoff = d
|
|
}
|
|
}
|
|
|
|
// Enqueue inserts a new outgoing email. notificationID may be empty. The
|
|
// email is immediately eligible for delivery (next_attempt_at = now).
|
|
func (q *Queue) Enqueue(ctx context.Context, notificationID, to, subject, body string) (string, error) {
|
|
if to == "" {
|
|
return "", fmt.Errorf("enqueue: to address is required")
|
|
}
|
|
if subject == "" {
|
|
return "", fmt.Errorf("enqueue: subject is required")
|
|
}
|
|
id, err := randomQueueID()
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
now := time.Now().UTC()
|
|
if _, err := q.db.ExecContext(ctx, `
|
|
INSERT INTO email_queue (id, notification_id, to_address, subject, body, status, attempts, max_attempts, next_attempt_at, created_at)
|
|
VALUES (?, ?, ?, ?, ?, 'pending', 0, 3, ?, ?)
|
|
`, id, nullableString(notificationID), to, subject, body, formatTime(now), formatTime(now)); err != nil {
|
|
return "", fmt.Errorf("enqueue email: %w", err)
|
|
}
|
|
return id, nil
|
|
}
|
|
|
|
// MarkNotificationEmailed stamps notifications.emailed_at for the given
|
|
// notification id. Called by the worker once the email is dispatched.
|
|
func (q *Queue) MarkNotificationEmailed(ctx context.Context, notificationID string) error {
|
|
if notificationID == "" {
|
|
return nil
|
|
}
|
|
_, err := q.db.ExecContext(ctx, `
|
|
UPDATE notifications SET emailed_at = ? WHERE id = ?
|
|
`, formatTime(time.Now().UTC()), notificationID)
|
|
if err != nil {
|
|
return fmt.Errorf("mark notification emailed: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// Run starts the worker loop. It blocks until ctx is cancelled.
|
|
func (q *Queue) Run(ctx context.Context) {
|
|
q.logger.Info("email queue worker started", "poll_interval", q.pollInterval)
|
|
ticker := time.NewTicker(q.pollInterval)
|
|
defer ticker.Stop()
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
q.logger.Info("email queue worker stopped")
|
|
return
|
|
case <-ticker.C:
|
|
if err := q.processBatch(ctx); err != nil {
|
|
q.logger.Warn("email queue batch failed", "error", err)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// processBatch claims and sends up to 25 pending emails per tick.
|
|
func (q *Queue) processBatch(ctx context.Context) error {
|
|
if q.sender == nil {
|
|
// No sender configured; leave rows pending so a future sender can
|
|
// pick them up once wired.
|
|
return nil
|
|
}
|
|
rows, err := q.claimBatch(ctx, 25)
|
|
if err != nil {
|
|
return fmt.Errorf("claim batch: %w", err)
|
|
}
|
|
for _, row := range rows {
|
|
if err := q.deliver(ctx, row); err != nil {
|
|
q.logger.Warn("deliver email failed", "id", row.ID, "error", err)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// claimBatch atomically leases pending rows by flipping their status to
|
|
// 'pending' but bumping attempts so a concurrent worker cannot re-claim
|
|
// them until the next_attempt_at expires. We use a single UPDATE with a
|
|
// subquery to keep this race-free under SQLite.
|
|
func (q *Queue) claimBatch(ctx context.Context, limit int) ([]QueueRow, error) {
|
|
tx, err := q.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("begin claim: %w", err)
|
|
}
|
|
defer func() { _ = tx.Rollback() }()
|
|
|
|
rows, err := tx.QueryContext(ctx, `
|
|
SELECT id, COALESCE(notification_id, ''), to_address, subject, body, status, attempts, max_attempts,
|
|
COALESCE(last_error, ''), next_attempt_at, COALESCE(sent_at, ''), created_at
|
|
FROM email_queue
|
|
WHERE status = 'pending' AND next_attempt_at <= ?
|
|
ORDER BY next_attempt_at ASC
|
|
LIMIT ?
|
|
`, formatTime(time.Now().UTC()), limit)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("query pending: %w", err)
|
|
}
|
|
|
|
var claimed []QueueRow
|
|
for rows.Next() {
|
|
row, err := scanQueueRow(rows)
|
|
if err != nil {
|
|
rows.Close()
|
|
return nil, err
|
|
}
|
|
claimed = append(claimed, row)
|
|
}
|
|
rows.Close()
|
|
if err := rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
for _, row := range claimed {
|
|
if _, err := tx.ExecContext(ctx, `
|
|
UPDATE email_queue SET attempts = attempts + 1 WHERE id = ?
|
|
`, row.ID); err != nil {
|
|
return nil, fmt.Errorf("lease row %s: %w", row.ID, err)
|
|
}
|
|
}
|
|
|
|
if err := tx.Commit(); err != nil {
|
|
return nil, fmt.Errorf("commit claim: %w", err)
|
|
}
|
|
return claimed, nil
|
|
}
|
|
|
|
// deliver attempts one email. On success it marks the row sent and stamps
|
|
// the linked notification. On failure it schedules a retry with exponential
|
|
// backoff or marks the row failed if attempts are exhausted.
|
|
func (q *Queue) deliver(ctx context.Context, row QueueRow) error {
|
|
err := q.sender.Send(ctx, []string{row.ToAddress}, row.Subject, row.Body)
|
|
if err == nil {
|
|
if _, err := q.db.ExecContext(ctx, `
|
|
UPDATE email_queue SET status = 'sent', sent_at = ?, last_error = '' WHERE id = ?
|
|
`, formatTime(time.Now().UTC()), row.ID); err != nil {
|
|
return fmt.Errorf("mark sent: %w", err)
|
|
}
|
|
if err := q.MarkNotificationEmailed(ctx, row.NotificationID); err != nil {
|
|
q.logger.Warn("mark notification emailed", "id", row.NotificationID, "error", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// Failure: retry or give up.
|
|
if row.Attempts >= row.MaxAttempts {
|
|
if _, ferr := q.db.ExecContext(ctx, `
|
|
UPDATE email_queue SET status = 'failed', last_error = ? WHERE id = ?
|
|
`, truncateErr(err.Error()), row.ID); ferr != nil {
|
|
return fmt.Errorf("mark failed: %w", ferr)
|
|
}
|
|
q.logger.Warn("email permanently failed", "id", row.ID, "attempts", row.Attempts, "error", err)
|
|
return nil
|
|
}
|
|
|
|
backoff := q.backoff(row.Attempts)
|
|
next := time.Now().UTC().Add(backoff)
|
|
if _, err := q.db.ExecContext(ctx, `
|
|
UPDATE email_queue SET status = 'pending', next_attempt_at = ?, last_error = ? WHERE id = ?
|
|
`, formatTime(next), truncateErr(err.Error()), row.ID); err != nil {
|
|
return fmt.Errorf("schedule retry: %w", err)
|
|
}
|
|
|
|
// TemporaryError (Postmark 5xx) is retried with the same backoff as
|
|
// other errors; the max-attempts cap still bounds total work.
|
|
q.logger.Info("email retry scheduled", "id", row.ID, "attempt", row.Attempts, "backoff", backoff, "error", err)
|
|
return nil
|
|
}
|
|
|
|
func (q *Queue) backoff(attempts int) time.Duration {
|
|
// 2^attempts seconds, capped: 2s, 4s, 8s, 16s, ... up to maxBackoff.
|
|
d := time.Duration(math.Pow(2, float64(attempts))) * time.Second
|
|
if d > q.maxBackoff {
|
|
d = q.maxBackoff
|
|
}
|
|
if d < time.Second {
|
|
d = time.Second
|
|
}
|
|
return d
|
|
}
|
|
|
|
// PendingCount returns the number of rows in pending status, primarily for
|
|
// tests and operator dashboards.
|
|
func (q *Queue) PendingCount(ctx context.Context) (int, error) {
|
|
var count int
|
|
err := q.db.QueryRowContext(ctx, `SELECT COUNT(1) FROM email_queue WHERE status = 'pending'`).Scan(&count)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
return count, nil
|
|
}
|
|
|
|
// ListFailed returns rows that have exhausted their retries, for operator
|
|
// inspection.
|
|
func (q *Queue) ListFailed(ctx context.Context, limit int) ([]QueueRow, error) {
|
|
if limit <= 0 {
|
|
limit = 50
|
|
}
|
|
rows, err := q.db.QueryContext(ctx, `
|
|
SELECT id, COALESCE(notification_id, ''), to_address, subject, body, status, attempts, max_attempts,
|
|
COALESCE(last_error, ''), next_attempt_at, COALESCE(sent_at, ''), created_at
|
|
FROM email_queue
|
|
WHERE status = 'failed'
|
|
ORDER BY created_at DESC
|
|
LIMIT ?
|
|
`, limit)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
|
|
var out []QueueRow
|
|
for rows.Next() {
|
|
row, err := scanQueueRow(rows)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
out = append(out, row)
|
|
}
|
|
return out, rows.Err()
|
|
}
|
|
|
|
func scanQueueRow(rows *sql.Rows) (QueueRow, error) {
|
|
var r QueueRow
|
|
var sentAt, lastError string
|
|
var nextAttempt, createdAt string
|
|
if err := rows.Scan(&r.ID, &r.NotificationID, &r.ToAddress, &r.Subject, &r.Body, &r.Status,
|
|
&r.Attempts, &r.MaxAttempts, &lastError, &nextAttempt, &sentAt, &createdAt); err != nil {
|
|
return QueueRow{}, fmt.Errorf("scan queue row: %w", err)
|
|
}
|
|
r.LastError = lastError
|
|
var err error
|
|
r.NextAttemptAt, err = time.Parse(time.RFC3339, nextAttempt)
|
|
if err != nil {
|
|
return QueueRow{}, fmt.Errorf("parse next_attempt_at: %w", err)
|
|
}
|
|
r.CreatedAt, err = time.Parse(time.RFC3339, createdAt)
|
|
if err != nil {
|
|
return QueueRow{}, fmt.Errorf("parse created_at: %w", err)
|
|
}
|
|
if sentAt != "" {
|
|
t, err := time.Parse(time.RFC3339, sentAt)
|
|
if err == nil {
|
|
r.SentAt = &t
|
|
}
|
|
}
|
|
return r, nil
|
|
}
|
|
|
|
func formatTime(t time.Time) string {
|
|
return t.UTC().Format(time.RFC3339)
|
|
}
|
|
|
|
func nullableString(s string) any {
|
|
if s == "" {
|
|
return nil
|
|
}
|
|
return s
|
|
}
|
|
|
|
func truncateErr(s string) string {
|
|
if len(s) > 512 {
|
|
return s[:512] + "..."
|
|
}
|
|
return s
|
|
}
|
|
|
|
// randomQueueID produces an id with enough entropy to avoid collisions. We
|
|
// keep a small helper in-package to avoid an import cycle on auth.
|
|
func randomQueueID() (string, error) {
|
|
id, err := randomToken(18)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
return "email:" + id, nil
|
|
}
|