mailgo/internal/outbound/manager.go

package outbound

import (
	"fmt"
	"log"
	"net/mail"
	"strings"
	"sync"
	"time"

	"mail_go/config"
	"mail_go/internal/db"
	"mail_go/internal/store"

	"github.com/google/uuid"
)

// Manager orchestrates the outbound delivery queue: enqueueing messages,
// concurrent background delivery workers, exponential backoff retries, DKIM
// signing, per-user rate limits and failure bounces.
//
// 并发模型：一个 dispatcher goroutine 周期性扫描队列并原子抢占（Claim）
// 待投递项，投递给 worker 池（默认 4 个 goroutine）并行发送；同一收件域
// （或中继）的连接数受 max_concurrent_per_domain 限制。workers=0/1 时退
// 化为串行投递（旧行为）。
type Manager struct {
	cfg      config.OutboundConfig
	hostname string // EHLO hostname
	mailer   *Mailer
	stores   *store.Stores

	kick  chan struct{}
	stop  chan struct{}
	done  chan struct{}
	once  sync.Once
	wg    sync.WaitGroup
	mu    sync.Mutex
	lim   map[uint]*userWindow
	jobs  chan *db.OutboundMessage
	domMu sync.Mutex
	dom   map[string]chan struct{} // 每域并发信号量

	// deliver 执行单封投递，默认走 m.mailer.Deliver；测试可注入替换。
	deliver func(from, to string, data []byte) (string, error)
}

// userWindow tracks a user's sending rate within fixed windows.
type userWindow struct {
	minuteStart time.Time
	minuteCount int
	dayStart    time.Time
	dayCount    int
}

// NewManager creates an outbound delivery Manager.
// hostname is the EHLO name presented to remote servers (defaults to "localhost").
func NewManager(cfg config.OutboundConfig, hostname string, stores *store.Stores) *Manager {
	batchSize := cfg.BatchSize
	if batchSize <= 0 {
		batchSize = 50
	}
	workers := cfg.Workers
	if workers <= 1 {
		workers = 1
	}
	m := &Manager{
		cfg:      cfg,
		hostname: hostname,
		mailer:   NewMailer(hostname, time.Duration(cfg.ConnectTimeout)*time.Second),
		stores:   stores,
		kick:     make(chan struct{}, 1),
		stop:     make(chan struct{}),
		done:     make(chan struct{}),
		lim:      make(map[uint]*userWindow),
		jobs:     make(chan *db.OutboundMessage, batchSize),
		dom:      make(map[string]chan struct{}),
	}
	m.deliver = m.mailer.Deliver
	m.mailer.IPFamily = cfg.IPFamily
	m.mailer.SourceIP = cfg.SourceIP
	if cfg.SourceIP != "" {
		log.Printf("outbound: binding source address %s (ip_family=%s)", cfg.SourceIP, cfg.IPFamily)
	}

	if cfg.RelayHost != "" {
		m.mailer.Relay = &RelayConfig{
			Host:        cfg.RelayHost,
			Port:        cfg.RelayPort,
			Username:    cfg.RelayUser,
			Password:    cfg.RelayPassword,
			StartTLS:    cfg.RelayStartTLS,
			TLSInsecure: cfg.RelayTLSInsecure,
		}
		log.Printf("outbound: using smarthost relay %s:%d", cfg.RelayHost, cfg.RelayPort)
	}
	log.Printf("outbound: %d delivery workers, batch=%d, per-domain concurrency=%d",
		workers, batchSize, cfg.MaxConcurrentPerDomain)
	return m
}

// Start launches the dispatcher and the delivery worker pool.
func (m *Manager) Start() {
	interval := time.Duration(m.cfg.PollInterval) * time.Second
	if interval <= 0 {
		interval = 15 * time.Second
	}

	// 调度者：启动时立即扫描一次（清空积压），之后按周期扫描 +
	// 原子抢占待投递项，投递给 worker 池。
	m.wg.Add(1)
	go func() {
		defer m.wg.Done()
		ticker := time.NewTicker(interval)
		defer ticker.Stop()
		log.Printf("outbound: dispatcher started (interval=%s, max_attempts=%d)", interval, m.cfg.MaxAttempts)
		m.processDue()
		for {
			select {
			case <-ticker.C:
				m.processDue()
			case <-m.kick:
				m.processDue()
			case <-m.stop:
				close(m.jobs)
				close(m.done)
				return
			}
		}
	}()

	// Worker 池：并行投递。
	workers := m.cfg.Workers
	if workers <= 1 {
		workers = 1
	}
	for i := 0; i < workers; i++ {
		m.wg.Add(1)
		go func() {
			defer m.wg.Done()
			for job := range m.jobs {
				m.deliverOne(job)
			}
		}()
	}
}

// Stop gracefully stops the dispatcher and workers.
func (m *Manager) Stop() {
	m.once.Do(func() {
		close(m.stop)
	})
	<-m.done
	m.wg.Wait()
}

// kickWorker nudges the worker to scan the queue immediately.
func (m *Manager) kickWorker() {
	select {
	case m.kick <- struct{}{}:
	default:
	}
}

// Enabled reports whether external delivery is configured on.
func (m *Manager) Enabled() bool {
	return m.cfg.MaxPerDay > 0
}

// MaxRecipients returns the maximum number of external recipients allowed
// per message (0 means unlimited).
func (m *Manager) MaxRecipients() int {
	return m.cfg.MaxRecipients
}

// Enqueue validates a sender/recipient pair, DKIM-signs the message once and
// stores it in the outbound queue for background delivery. The recipient must
// NOT be a local address — callers decide local vs external routing.
// Returns a permanent-style error for invalid input or rate-limit violations.
func (m *Manager) Enqueue(senderUser *db.User, from, to string, raw []byte) (*db.OutboundMessage, error) {
	if !m.Enabled() {
		return nil, fmt.Errorf("外部投递未启用")
	}

	to = strings.TrimSpace(to)
	addr, err := mail.ParseAddress(to)
	if err != nil {
		return nil, fmt.Errorf("收件人地址无效: %s", to)
	}
	to = addr.Address

	at := strings.LastIndex(to, "@")
	if at < 0 || at == len(to)-1 {
		return nil, fmt.Errorf("收件人地址无效: %s", to)
	}
	recipientDom := strings.ToLower(to[at+1:])

	// Local addresses must never enter the outbound queue.
	if _, err := m.stores.Users.GetByEmail(to); err == nil {
		return nil, fmt.Errorf("收件人 %s 是本地地址，应走本地投递", to)
	}

	// Rate limiting per sender user.
	if senderUser != nil {
		if err := m.checkRateLimit(senderUser.ID); err != nil {
			return nil, err
		}
	}

	// DKIM-sign once with the sender domain's key.
	signed, err := m.signForSender(from, raw)
	if err != nil {
		log.Printf("outbound: DKIM signing failed for %s: %v", from, err)
		signed = raw
	}

	item := &db.OutboundMessage{
		MessageID:     fmt.Sprintf("<%s@outbound>", uuid.New().String()),
		UserID:        userIDOrZero(senderUser),
		FromAddr:      from,
		ToAddr:        to,
		RecipientDom:  recipientDom,
		RawData:       string(signed),
		Status:        db.OutboundStatusPending,
		Attempts:      0,
		NextAttemptAt: time.Now(),
	}
	if err := m.stores.Outbound.Create(item); err != nil {
		return nil, fmt.Errorf("写入外发队列失败: %w", err)
	}

	log.Printf("outbound: queued %s -> %s (id=%d)", from, to, item.ID)
	m.kickWorker()
	return item, nil
}

func userIDOrZero(u *db.User) uint {
	if u == nil {
		return 0
	}
	return u.ID
}

// signForSender looks up the sender domain's DKIM key and signs the message.
func (m *Manager) signForSender(from string, raw []byte) ([]byte, error) {
	at := strings.LastIndex(from, "@")
	if at < 0 || at == len(from)-1 {
		return raw, fmt.Errorf("无效发件人地址: %s", from)
	}
	domName := strings.ToLower(from[at+1:])

	domain, err := m.stores.Domains.GetByName(domName)
	if err != nil {
		return raw, nil // domain not managed locally; send unsigned
	}
	return SignDKIM(raw, domain.Name, domain.DkimSelector, domain.DkimPrivateKey)
}

// checkRateLimit enforces per-user minute/day sending limits.
func (m *Manager) checkRateLimit(userID uint) error {
	m.mu.Lock()
	defer m.mu.Unlock()

	now := time.Now()
	w := m.lim[userID]
	if w == nil || now.Sub(w.dayStart) >= 24*time.Hour {
		w = &userWindow{minuteStart: now, dayStart: now}
		m.lim[userID] = w
	} else if now.Sub(w.minuteStart) >= time.Minute {
		w.minuteStart = now
		w.minuteCount = 0
	}

	if m.cfg.MaxPerMin > 0 && w.minuteCount >= m.cfg.MaxPerMin {
		log.Printf("outbound: rate limit exceeded (per minute) for user %d", userID)
		return fmt.Errorf("发送频率超限：每分钟最多 %d 封", m.cfg.MaxPerMin)
	}
	if m.cfg.MaxPerDay > 0 && w.dayCount >= m.cfg.MaxPerDay {
		log.Printf("outbound: rate limit exceeded (per day) for user %d", userID)
		return fmt.Errorf("发送频率超限：每日最多 %d 封", m.cfg.MaxPerDay)
	}

	w.minuteCount++
	w.dayCount++
	return nil
}

// processDue scans the queue and dispatches due items to the worker pool.
// Each item is atomically claimed (status -> sending) before dispatch so
// that concurrent workers never deliver the same message twice. When all
// workers are busy the dispatcher blocks here, naturally throttling claims;
// remaining due items are picked up on the next scan.
func (m *Manager) processDue() {
	batchSize := m.cfg.BatchSize
	if batchSize <= 0 {
		batchSize = 50
	}
	items, err := m.stores.Outbound.ListDue(time.Now(), batchSize)
	if err != nil {
		log.Printf("outbound: loading due queue failed: %v", err)
		return
	}
	for i := range items {
		claimed, err := m.stores.Outbound.Claim(items[i].ID)
		if err != nil {
			log.Printf("outbound: claim item %d failed: %v", items[i].ID, err)
			continue
		}
		if !claimed {
			// 已被其他调度周期抢占（并发下应不会发生，防御性跳过）
			continue
		}
		item := items[i]
		item.Status = db.OutboundStatusSending
		m.jobs <- &item
	}
}

// acquireDomain 获取收件域（或中继）的并发信号量，限制对同一目标域同时
// 打开的 SMTP 连接数。limit <= 0 表示不限制。
func (m *Manager) acquireDomain(domain string) func() {
	limit := m.cfg.MaxConcurrentPerDomain
	if limit <= 0 {
		return func() {}
	}

	m.domMu.Lock()
	sem := m.dom[domain]
	if sem == nil {
		sem = make(chan struct{}, limit)
		m.dom[domain] = sem
	}
	m.domMu.Unlock()

	sem <- struct{}{}
	return func() { <-sem }
}

// deliverKey 返回并发限制使用的目标标识：配置了中继时所有连接都打向同一
// smarthost，统一按 "relay" 限制；否则按收件域名限制。
func (m *Manager) deliverKey(to string) string {
	if m.mailer.Relay != nil && m.mailer.Relay.Host != "" {
		return "relay"
	}
	at := strings.LastIndex(to, "@")
	if at < 0 || at == len(to)-1 {
		return ""
	}
	return strings.ToLower(to[at+1:])
}

// deliverOne performs a single delivery attempt for a queue item.
// 调用前该项已被原子抢占为 sending，此处不再重复置位。
func (m *Manager) deliverOne(item *db.OutboundMessage) {
	release := m.acquireDomain(item.ToAddr)
	defer release()

	resp, err := m.deliver(item.FromAddr, item.ToAddr, []byte(item.RawData))

	now := time.Now()
	item.Attempts++

	if err == nil {
		item.Status = db.OutboundStatusSent
		item.LastResponse = resp
		item.LastError = ""
		item.CompletedAt = &now
		if saveErr := m.stores.Outbound.Update(item); saveErr != nil {
			log.Printf("outbound: update item %d to sent failed: %v", item.ID, saveErr)
			return
		}
		log.Printf("outbound: delivered %s -> %s (id=%d, attempts=%d)", item.FromAddr, item.ToAddr, item.ID, item.Attempts)
		return
	}

	var de *DeliveryError
	permanent := false
	if ok := asDeliveryError(err, &de); ok {
		permanent = de.Permanent
	}
	item.LastResponse = ""
	item.LastError = err.Error()

	if permanent || item.Attempts >= m.cfg.MaxAttempts {
		item.Status = db.OutboundStatusFailed
		item.CompletedAt = &now
		if saveErr := m.stores.Outbound.Update(item); saveErr != nil {
			log.Printf("outbound: update item %d to failed failed: %v", item.ID, saveErr)
			return
		}
		log.Printf("outbound: permanent failure %s -> %s (id=%d): %v", item.FromAddr, item.ToAddr, item.ID, err)
		m.bounce(item, err)
		return
	}

	// Temporary failure: exponential backoff retry.
	item.Status = db.OutboundStatusDeferred
	backoff := time.Duration(m.cfg.RetryBaseMin) * time.Minute
	backoff <<= (item.Attempts - 1)
	if backoff > 24*time.Hour {
		backoff = 24 * time.Hour
	}
	item.NextAttemptAt = now.Add(backoff)
	if saveErr := m.stores.Outbound.Update(item); saveErr != nil {
		log.Printf("outbound: update item %d to deferred failed: %v", item.ID, saveErr)
		return
	}
	log.Printf("outbound: temporary failure %s -> %s (id=%d, attempt=%d, retry in %s): %v",
		item.FromAddr, item.ToAddr, item.ID, item.Attempts, backoff, err)
}

// bounce delivers a non-delivery notice to the sender's INBOX.
func (m *Manager) bounce(item *db.OutboundMessage, deliveryErr error) {
	sender, err := m.stores.Users.GetByEmail(item.FromAddr)
	if err != nil {
		log.Printf("outbound: cannot bounce %s: sender is not a local user", item.FromAddr)
		return
	}

	now := time.Now()
	postmaster := "Mail Delivery System <postmaster@" + m.hostname + ">"
	subject := fmt.Sprintf("邮件投递失败: %s", item.ToAddr)

	body := "这是一封系统退信通知。\r\n\r\n" +
		fmt.Sprintf("您的邮件未能投递到以下收件人：\r\n\r\n  收件人：%s\r\n  失败原因：%s\r\n  投递时间：%s\r\n  尝试次数：%d\r\n\r\n",
			item.ToAddr, deliveryErr.Error(), now.Format("2006-01-02 15:04:05"), item.Attempts) +
		"如果收件人地址无误，请稍后重试；连续失败可能表示收件地址不存在或对方服务器拒收。\r\n"

	msg := &db.Message{
		UserID:    sender.ID,
		MessageID: fmt.Sprintf("<bounce-%d@%s>", item.ID, m.hostname),
		Folder:    "INBOX",
		FromAddr:  postmaster,
		ToAddr:    item.FromAddr,
		Subject:   subject,
		TextBody:  body,
		Date:      now,
		IsRead:    false,
	}
	if err := m.stores.Mails.Create(msg); err != nil {
		log.Printf("outbound: bounce message creation failed: %v", err)
		return
	}
	log.Printf("outbound: bounce delivered to %s for failed delivery of %s", item.FromAddr, item.ToAddr)
}

// Retry resets a failed/deferred queue item for immediate redelivery.
func (m *Manager) Retry(id uint) error {
	item, err := m.stores.Outbound.GetByID(id)
	if err != nil {
		return err
	}
	item.Status = db.OutboundStatusPending
	item.Attempts = 0
	item.LastError = ""
	item.LastResponse = ""
	item.CompletedAt = nil
	item.NextAttemptAt = time.Now()
	if err := m.stores.Outbound.Update(item); err != nil {
		return err
	}
	m.kickWorker()
	return nil
}

// Cancel marks a queue item as canceled by the administrator.
func (m *Manager) Cancel(id uint) error {
	item, err := m.stores.Outbound.GetByID(id)
	if err != nil {
		return err
	}
	if item.Status == db.OutboundStatusSent {
		return fmt.Errorf("已送达的邮件无法取消")
	}
	now := time.Now()
	item.Status = db.OutboundStatusCanceled
	item.LastError = "管理员取消"
	item.CompletedAt = &now
	return m.stores.Outbound.Update(item)
}

// asDeliveryError extracts a *DeliveryError from an error chain.
func asDeliveryError(err error, target **DeliveryError) bool {
	for err != nil {
		if de, ok := err.(*DeliveryError); ok {
			*target = de
			return true
		}
		type unwrapper interface{ Unwrap() error }
		u, ok := err.(unwrapper)
		if !ok {
			return false
		}
		err = u.Unwrap()
	}
	return false
}

// StatusText returns a human-readable Chinese label for a queue status.
func StatusText(status string) string {
	switch status {
	case db.OutboundStatusPending:
		return "待发送"
	case db.OutboundStatusSending:
		return "发送中"
	case db.OutboundStatusSent:
		return "已送达"
	case db.OutboundStatusDeferred:
		return "等待重试"
	case db.OutboundStatusFailed:
		return "失败"
	case db.OutboundStatusCanceled:
		return "已取消"
	default:
		return status
	}
}