mailgoserver/internal/relay/queue.go

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, permanent := 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 !permanent && 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, or the remote server gave a permanent (RFC 5321 5xx) rejection
	// on the very first attempt — either way this is final. 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 this runs, 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)
	}
}