gomail/internal/queue/queue.go

// Package queue implements the outbound delivery worker: polls due entries,
// resolves MX records, delivers via net/smtp (stdlib), and handles retry
// backoff and bounce generation for permanent failures.
package queue

import (
	"context"
	"crypto/tls"
	"encoding/json"
	"fmt"
	"log/slog"
	"net"
	"net/smtp"
	"strings"
	"time"

	"gomail/internal/dane"
	"gomail/internal/db"
	"gomail/internal/dkim"
	"gomail/internal/mailstore"
	"gomail/internal/mtasts"
)

const (
	maxAttempts   = 5
	pollInterval  = 30 * time.Second
	deliveryTimeout = 60 * time.Second
)

// Deliverer is the interface the worker uses to actually hand a message to a
// remote MTA — abstracted so tests can inject a fake without real network
// access (outbound port 25 is blocked in most sandboxed/dev environments).
type Deliverer interface {
	Deliver(from, to string, raw []byte) error
}

// KeyLookup resolves the DKIM signing key for a sending domain, returning
// (privateKeyPEM, selector, found). The worker calls this fresh on every
// delivery attempt (not cached at startup) so key rotation via the admin
// portal takes effect immediately without a restart.
type KeyLookup func(fromDomain string) (privateKeyPEM []byte, selector string, ok bool)

// Worker polls outbound_queue and processes due entries.
type Worker struct {
	database  *db.DB
	store     *mailstore.Store
	deliverer Deliverer
	keyLookup KeyLookup
	stopCh    chan struct{}
}

func NewWorker(database *db.DB, store *mailstore.Store) *Worker {
	return &Worker{
		database:  database,
		store:     store,
		deliverer: &MXDeliverer{Hostname: "gomail", Database: database},
		stopCh:    make(chan struct{}),
	}
}

// WithDeliverer overrides the delivery mechanism — used by tests.
func (w *Worker) WithDeliverer(d Deliverer) *Worker {
	w.deliverer = d
	return w
}

// WithKeyLookup enables DKIM signing before every delivery attempt. Signing
// happens here in the worker — not inside a specific Deliverer implementation
// — so it applies uniformly regardless of transport (MX delivery, a test
// fake, or any future alternative).
func (w *Worker) WithKeyLookup(kl KeyLookup) *Worker {
	w.keyLookup = kl
	return w
}

// Run starts the polling loop. Blocks until Stop is called.
func (w *Worker) Run() {
	ticker := time.NewTicker(pollInterval)
	defer ticker.Stop()

	slog.Info("outbound queue worker started", "poll_interval", pollInterval)
	w.ProcessOnce() // run immediately on start, don't wait for the first tick

	for {
		select {
		case <-w.stopCh:
			return
		case <-ticker.C:
			w.ProcessOnce()
		}
	}
}

func (w *Worker) Stop() {
	close(w.stopCh)
}

// ProcessOnce runs a single pass: attempts delivery for all due entries,
// then bounces anything that has exhausted its retry budget.
func (w *Worker) ProcessOnce() {
	entries, err := w.database.DueOutboundEntries(maxAttempts, 100)
	if err != nil {
		slog.Error("queue: failed to load due entries", "err", err)
		return
	}

	for _, entry := range entries {
		w.attemptDelivery(entry)
	}

	failed, err := w.database.PermanentlyFailedEntries(maxAttempts)
	if err != nil {
		slog.Error("queue: failed to load permanently failed entries", "err", err)
		return
	}
	for _, entry := range failed {
		w.bounce(entry)
	}
}

func (w *Worker) attemptDelivery(entry db.OutboundQueueEntry) {
	raw, err := w.store.Read(entry.EMLPath)
	if err != nil {
		slog.Error("queue: failed to read queued message", "id", entry.ID, "err", err)
		w.scheduleRetry(entry, fmt.Sprintf("read failed: %v", err))
		return
	}

	if w.keyLookup != nil {
		fromDomain := domainOf(entry.FromAddress)
		if privateKeyPEM, selector, ok := w.keyLookup(fromDomain); ok {
			signed, err := dkim.Sign(privateKeyPEM, fromDomain, selector, raw)
			if err != nil {
				slog.Warn("queue: DKIM signing failed, sending unsigned", "domain", fromDomain, "err", err)
			} else {
				raw = signed
			}
		}
	}

	err = w.deliverer.Deliver(entry.FromAddress, entry.ToAddress, raw)
	if err == nil {
		slog.Info("queue: delivered", "to", entry.ToAddress, "attempts", entry.Attempts+1)
		if delErr := w.database.DeleteOutboundEntry(entry.ID); delErr != nil {
			slog.Error("queue: failed to delete completed entry", "err", delErr)
		}
		return
	}

	if isPermanentError(err) {
		slog.Warn("queue: permanent delivery failure, will bounce", "to", entry.ToAddress, "err", err)
		// Fast-forward attempts to the max so the next ProcessOnce pass bounces
		// it immediately, instead of waiting through the full retry schedule.
		remaining := maxAttempts - entry.Attempts
		for i := 0; i < remaining; i++ {
			w.database.RetryOutboundEntry(entry.ID, time.Now().UTC(), err.Error())
		}
		return
	}

	slog.Info("queue: temporary delivery failure, will retry", "to", entry.ToAddress, "attempt", entry.Attempts+1, "err", err)
	w.scheduleRetry(entry, err.Error())
}

func (w *Worker) scheduleRetry(entry db.OutboundQueueEntry, errMsg string) {
	backoff := backoffDuration(entry.Attempts + 1)
	next := time.Now().UTC().Add(backoff)
	if err := w.database.RetryOutboundEntry(entry.ID, next, errMsg); err != nil {
		slog.Error("queue: failed to schedule retry", "err", err)
	}
}

// backoffDuration implements exponential backoff: 5m, 20m, 1h20m, 5h20m, ~21h
// for attempts 1 through 5, capping the total retry window near 5 days as
// planned (RFC 5321 recommends retrying for at least 4-5 days before giving up).
func backoffDuration(attempt int) time.Duration {
	base := 5 * time.Minute
	d := base
	for i := 1; i < attempt; i++ {
		d *= 4
	}
	max := 24 * time.Hour
	if d > max {
		d = max
	}
	return d
}

// bounce generates a DSN-style bounce message and delivers it to the local
// sender's INBOX (the original MAIL FROM on submission is always a local
// user, since session.go enforces that match at RCPT TO time).
func (w *Worker) bounce(entry db.OutboundQueueEntry) {
	user, err := w.database.LookupUserByEmail(entry.FromAddress)
	if err != nil {
		slog.Error("queue: cannot bounce — original sender not found locally", "from", entry.FromAddress, "err", err)
		w.database.DeleteOutboundEntry(entry.ID)
		return
	}

	bounceBody := fmt.Sprintf(
		"From: Mail Delivery System <postmaster@%s>\r\n"+
			"To: %s\r\n"+
			"Subject: Undelivered Mail Returned to Sender\r\n"+
			"Date: %s\r\n"+
			"\r\n"+
			"This is an automatically generated Delivery Status Notification.\r\n\r\n"+
			"Delivery to the following recipient failed permanently after %d attempts:\r\n\r\n"+
			"    %s\r\n\r\n"+
			"Last error: %s\r\n\r\n"+
			"This is the final notification; no further attempts will be made.\r\n",
		domainOf(entry.FromAddress), entry.FromAddress, time.Now().UTC().Format(time.RFC1123Z),
		entry.Attempts, entry.ToAddress, entry.LastError,
	)

	if _, err := w.store.Deliver(user.ID, user.Email, "INBOX", []byte(bounceBody)); err != nil {
		slog.Error("queue: failed to deliver bounce", "err", err)
		return
	}

	slog.Info("queue: bounce delivered", "to", entry.FromAddress, "original_recipient", entry.ToAddress)
	w.database.DeleteOutboundEntry(entry.ID)
}

func domainOf(email string) string {
	parts := strings.SplitN(email, "@", 2)
	if len(parts) == 2 {
		return parts[1]
	}
	return "localhost"
}

// isPermanentError distinguishes 5xx (permanent) from 4xx/network (temporary)
// SMTP failures — net/smtp wraps the server's textual response in the error,
// so we inspect it for the leading status code digit.
func isPermanentError(err error) bool {
	msg := err.Error()
	// net/smtp errors look like "553 5.1.1 User unknown" when they come from
	// the remote server's response.
	for _, code := range []string{"550", "551", "552", "553", "554"} {
		if strings.Contains(msg, code) {
			return true
		}
	}
	return false
}

// ── MX-resolving deliverer (stdlib net/smtp + net.LookupMX) ────────────────────

// MXDeliverer is the real production Deliverer: resolves the recipient
// domain's MX records, connects (with STARTTLS if offered), and hands off
// via net/smtp — Go's standard library SMTP client, chosen specifically to
// stay dependency-free for outbound delivery just as the inbound server is
// hand-rolled from net.Listener. Pure transport — DKIM signing (if any)
// happens in Worker.attemptDelivery before Deliver is called, so it applies
// uniformly regardless of which Deliverer implementation is in use.
type MXDeliverer struct {
	Hostname string // EHLO identity
	Database *db.DB // caches MTA-STS policies (RFC 8461 requires honoring max_age); nil just disables caching, not the feature — policy is re-fetched every delivery instead
}

func (d *MXDeliverer) Deliver(from, to string, raw []byte) error {
	domain := domainOf(to)
	mxHosts, err := lookupMXHosts(domain)
	if err != nil {
		return fmt.Errorf("451 4.4.3 MX lookup failed for %s: %w", domain, err)
	}

	var lastErr error
	for _, host := range mxHosts {
		if err := d.deliverToHost(host, from, to, raw); err != nil {
			lastErr = err
			continue
		}
		return nil
	}
	return lastErr
}

// tlsPolicy is what resolveTLSPolicy decides for one delivery attempt.
// mandatory distinguishes "TLS must succeed or this attempt fails" (DANE,
// MTA-STS enforce) from today's original opportunistic behavior (try,
// log and continue in plaintext on failure).
type tlsPolicy struct {
	config    *tls.Config
	mandatory bool
	source    string // for logging: "dane" | "mta-sts:enforce" | "opportunistic"
}

// resolveTLSPolicy decides what TLS behavior this delivery attempt must
// follow, most-specific first: DANE (per-host) over MTA-STS (per-domain)
// over plain opportunistic STARTTLS — today's original, unchanged default.
// A non-nil error means delivery to this host must not proceed at all
// (e.g. an MTA-STS enforce policy that doesn't list this host as valid —
// returned as a 550 so it bounces via the existing permanent-failure path
// instead of retrying forever against a host the domain's own policy
// disowns). See internal/dane and internal/mtasts package docs for the
// respective security models and caveats (DANE's DNSSEC dependency, in
// particular).
func (d *MXDeliverer) resolveTLSPolicy(ctx context.Context, host, domain string) (tlsPolicy, error) {
	opportunistic := tlsPolicy{
		config:    &tls.Config{ServerName: host, MinVersion: tls.VersionTLS12},
		mandatory: false,
		source:    "opportunistic",
	}

	if records, err := dane.Lookup(ctx, host, 25); err != nil {
		slog.Warn("DANE lookup failed, falling back to MTA-STS/opportunistic TLS", "host", host, "err", err)
	} else if len(records) > 0 {
		return tlsPolicy{
			config: &tls.Config{
				ServerName: host,
				MinVersion: tls.VersionTLS12,
				// Not disabling verification — replacing it. VerifyPeerCertificate
				// is DANE's own check (RFC 6698 usage 2/3), see internal/dane.
				InsecureSkipVerify:    true,
				VerifyPeerCertificate: dane.VerifyPeerCertificate(records),
			},
			mandatory: true,
			source:    "dane",
		}, nil
	}

	policy, err := d.mtaSTSPolicyFor(ctx, domain)
	if err != nil {
		slog.Warn("MTA-STS policy lookup failed, falling back to opportunistic TLS", "domain", domain, "err", err)
		return opportunistic, nil
	}
	if policy == nil || policy.Mode == "none" {
		return opportunistic, nil
	}

	hostMatches := false
	for _, pattern := range policy.MXPatterns {
		if mtasts.Matches(pattern, host) {
			hostMatches = true
			break
		}
	}

	switch policy.Mode {
	case "enforce":
		if !hostMatches {
			return tlsPolicy{}, fmt.Errorf("550 5.7.5 host %s is not listed in %s's MTA-STS policy (enforce mode)", host, domain)
		}
		return tlsPolicy{config: &tls.Config{ServerName: host, MinVersion: tls.VersionTLS12}, mandatory: true, source: "mta-sts:enforce"}, nil
	case "testing":
		if !hostMatches {
			slog.Warn("MTA-STS testing mode: host is not in the domain's policy (would fail under enforce)", "host", host, "domain", domain)
		}
		return opportunistic, nil
	default:
		return opportunistic, nil
	}
}

// mtaSTSPolicyFor returns domain's MTA-STS policy, using the DB cache when
// present and unexpired (RFC 8461 requires honoring the policy's own
// max_age) and fetching fresh otherwise. Returns (nil, nil) if the domain
// has no MTA-STS policy at all.
func (d *MXDeliverer) mtaSTSPolicyFor(ctx context.Context, domain string) (*mtasts.Policy, error) {
	if d.Database != nil {
		if cached, err := d.Database.GetMTASTSPolicy(domain); err == nil && time.Now().UTC().Before(cached.ExpiresAt) {
			var patterns []string
			if err := json.Unmarshal([]byte(cached.MXPatterns), &patterns); err == nil {
				return &mtasts.Policy{ID: cached.PolicyID, Mode: cached.Mode, MXPatterns: patterns, MaxAge: time.Duration(cached.MaxAge) * time.Second}, nil
			}
		}
	}

	policy, err := mtasts.Discover(ctx, domain)
	if err != nil || policy == nil {
		return policy, err
	}

	if d.Database != nil {
		patternsJSON, _ := json.Marshal(policy.MXPatterns)
		now := time.Now().UTC()
		if err := d.Database.UpsertMTASTSPolicy(&db.MTASTSPolicy{
			Domain: domain, PolicyID: policy.ID, Mode: policy.Mode,
			MXPatterns: string(patternsJSON), MaxAge: int(policy.MaxAge.Seconds()),
			FetchedAt: now, ExpiresAt: now.Add(policy.MaxAge),
		}); err != nil {
			slog.Warn("failed to cache MTA-STS policy", "domain", domain, "err", err)
		}
	}
	return policy, nil
}

func (d *MXDeliverer) deliverToHost(host, from, to string, raw []byte) error {
	domain := domainOf(to)

	conn, err := net.DialTimeout("tcp", host+":25", deliveryTimeout)
	if err != nil {
		return fmt.Errorf("421 4.4.1 connect to %s failed: %w", host, err)
	}
	defer conn.Close()
	conn.SetDeadline(time.Now().Add(deliveryTimeout))

	client, err := smtp.NewClient(conn, host)
	if err != nil {
		return fmt.Errorf("421 4.4.1 SMTP handshake with %s failed: %w", host, err)
	}
	defer client.Close()

	if err := client.Hello(d.Hostname); err != nil {
		return fmt.Errorf("EHLO to %s failed: %w", host, err)
	}

	policyCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	policy, err := d.resolveTLSPolicy(policyCtx, host, domain)
	cancel()
	if err != nil {
		return err
	}

	if ok, _ := client.Extension("STARTTLS"); ok {
		if err := client.StartTLS(policy.config); err != nil {
			if policy.mandatory {
				return fmt.Errorf("450 4.7.5 mandatory TLS (%s) failed for %s: %w", policy.source, host, err)
			}
			slog.Warn("STARTTLS failed, continuing without encryption", "host", host, "err", err)
		}
	} else if policy.mandatory {
		return fmt.Errorf("450 4.7.5 mandatory TLS (%s) required for %s but server does not offer STARTTLS", policy.source, host)
	}

	if err := client.Mail(from); err != nil {
		return err // preserves the remote server's status code in the error text
	}
	if err := client.Rcpt(to); err != nil {
		return err
	}

	w, err := client.Data()
	if err != nil {
		return err
	}
	if _, err := w.Write(raw); err != nil {
		return err
	}
	if err := w.Close(); err != nil {
		return err
	}

	return client.Quit()
}

func lookupMXHosts(domain string) ([]string, error) {
	mxs, err := net.LookupMX(domain)
	if err != nil || len(mxs) == 0 {
		// RFC 5321 §5.1 fallback: if no MX records, try the domain's A record directly.
		if _, aErr := net.LookupHost(domain); aErr == nil {
			return []string{domain}, nil
		}
		return nil, fmt.Errorf("no MX or A record for %s: %w", domain, err)
	}
	hosts := make([]string, len(mxs))
	for i, mx := range mxs {
		hosts[i] = strings.TrimSuffix(mx.Host, ".")
	}
	return hosts, nil
}