added email queue

internal/relay/queue.go

@@ -0,0 +1,129 @@
+package relay
+
+import (
+	"sync"
+	"time"
+
+	"mailgoserver/internal/db"
+)
+
+// retrySchedule is the fixed backoff between delivery attempts (five tries total, ~17h
+// worst case before giving up) — fixed, not admin-configurable, matching the
+// auto-reply loop-prevention window's own "fixed, keeps this simple" precedent
+// elsewhere in this codebase.
+var retrySchedule = []time.Duration{time.Minute, 15 * time.Minute, time.Hour, 4 * time.Hour, 12 * time.Hour}
+
+// ProcessQueueOnce claims up to maxBatch due esrv_relay_queue rows and attempts
+// delivery for each with up to maxConcurrent goroutines running at once — bounded so a
+// large backlog can never spawn unbounded outbound connections. Meant to be called on
+// a ticker (see main.go's runRelayQueueWorker); a no-op when nothing is due.
+func (r *Relay) ProcessQueueOnce(maxConcurrent, maxBatch int) {
+	items, err := r.DB.ClaimDueRelayQueueItems(maxBatch)
+	if err != nil {
+		r.Logger.Error("relay queue: claim due items: %v", err)
+		return
+	}
+	if len(items) == 0 {
+		return
+	}
+
+	sem := make(chan struct{}, maxConcurrent)
+	var wg sync.WaitGroup
+	for _, item := range items {
+		wg.Add(1)
+		sem <- struct{}{}
+		go func(item db.RelayQueueItem) {
+			defer wg.Done()
+			defer func() { <-sem }()
+			r.processQueueItem(item)
+		}(item)
+	}
+	wg.Wait()
+}
+
+// processQueueItem attempts one recipient-domain-group's delivery via the existing
+// deliverToDomain (same MX-iteration/STARTTLS/MTA-STS logic RelayEmailAsync's
+// synchronous path already uses — this is the only new call site, not a duplicate
+// implementation). On success or final failure, updates the placeholder "queued"
+// recipient-log rows (written by Session.Data at accept time) to their real outcome
+// and removes the queue row. On a failure with retries remaining, reschedules instead.
+func (r *Relay) processQueueItem(item db.RelayQueueItem) {
+	rcpts := make([]string, len(item.Recipients))
+	for i, rec := range item.Recipients {
+		rcpts[i] = rec.Recipient
+	}
+	status, serverResp, errCode, errMsg := r.deliverToDomain(item.Domain, item.MailFrom, rcpts, item.Content)
+
+	if status == "success" {
+		for _, rec := range item.Recipients {
+			if err := r.DB.UpdateEmailRecipientLogStatus(item.EmailLogID, rec.Recipient, "success", "", "", serverResp); err != nil {
+				r.Logger.Error("relay queue: update recipient log for %s: %v", rec.Recipient, err)
+			}
+		}
+		if err := r.DB.DeleteRelayQueueItem(item.ID); err != nil {
+			r.Logger.Error("relay queue: delete item %d: %v", item.ID, err)
+		}
+		r.finalizeEmailLogStatusIfDone(item.EmailLogID)
+		return
+	}
+
+	if item.Attempts < len(retrySchedule) {
+		next := time.Now().Add(retrySchedule[item.Attempts])
+		if err := r.DB.RescheduleRelayQueueItem(item.ID, next, errMsg); err != nil {
+			r.Logger.Error("relay queue: reschedule item %d: %v", item.ID, err)
+		}
+		return
+	}
+
+	// Retries exhausted — final failure. Resolve the recipient rows, drop the queue
+	// row, and bounce back to the original sender (skipped for a null sender, the
+	// same rule Session.Data's own immediate-bounce path already follows — replying
+	// to a bounce is the classic loop bug). Unlike that synchronous path, there's no
+	// live peer connection here to gate the bounce on IsIPBlacklisted — by the time
+	// retries are exhausted (minutes to hours later), this is no longer a live
+	// mailbox-enumeration oracle the way an instant response would be.
+	var failedResults []Result
+	for _, rec := range item.Recipients {
+		if err := r.DB.UpdateEmailRecipientLogStatus(item.EmailLogID, rec.Recipient, "failed", errCode, errMsg, serverResp); err != nil {
+			r.Logger.Error("relay queue: update recipient log for %s: %v", rec.Recipient, err)
+		}
+		failedResults = append(failedResults, Result{Recipient: rec.Recipient, RecipientType: rec.Type, Status: "failed", ErrorCode: errCode, ErrorMessage: errMsg, ServerResponse: serverResp})
+	}
+	if err := r.DB.DeleteRelayQueueItem(item.ID); err != nil {
+		r.Logger.Error("relay queue: delete item %d: %v", item.ID, err)
+	}
+	if item.MailFrom != "" {
+		if err := r.SendBounce(item.MailFrom, "", "", failedResults); err != nil {
+			r.Logger.Error("relay queue: send bounce to %s: %v", item.MailFrom, err)
+		}
+	}
+	r.finalizeEmailLogStatusIfDone(item.EmailLogID)
+}
+
+// finalizeEmailLogStatusIfDone recomputes and stores the parent email_log's overall
+// status once every domain-group for it has resolved (no esrv_relay_queue rows left
+// referencing it) — recomputed from the now-current recipient-log rows via the same
+// overallStatus logic LogEmail used at accept time, so a message that started
+// "queued" naturally settles into relayed/partial/failed once nothing's left pending.
+func (r *Relay) finalizeEmailLogStatusIfDone(emailLogID int64) {
+	pending, err := r.DB.CountPendingRelayQueueItemsForEmailLog(emailLogID)
+	if err != nil {
+		r.Logger.Error("relay queue: count pending for email_log %d: %v", emailLogID, err)
+		return
+	}
+	if pending > 0 {
+		return
+	}
+	recipients, err := r.DB.ListRecipientLogsForEmail(emailLogID)
+	if err != nil {
+		r.Logger.Error("relay queue: list recipients for email_log %d: %v", emailLogID, err)
+		return
+	}
+	results := make([]Result, len(recipients))
+	for i, rec := range recipients {
+		results[i] = Result{Status: rec.Status}
+	}
+	if err := r.DB.UpdateEmailLogStatus(emailLogID, overallStatus(results)); err != nil {
+		r.Logger.Error("relay queue: update email_log %d status: %v", emailLogID, err)
+	}
+}