NewSzxcn-Email/apps/api/internal/app/send_queue.go

package app

import (
	"context"
	"database/sql"
	"encoding/base64"
	"errors"
	"os"
	"path/filepath"
	"strings"
	"time"
)

const (
	sendQueueStatusQueued    = "queued"
	sendQueueStatusSending   = "sending"
	sendQueueStatusDelivered = "delivered"
	sendQueueStatusFailed    = "failed"
	sendQueueStatusCanceled  = "canceled"

	sendAuditAccepted  = "accepted"
	sendAuditQueued    = "queued"
	sendAuditDelivered = "delivered"
	sendAuditFailed    = "failed"
	sendAuditRetry     = "retry"
	sendAuditCanceled  = "canceled"

	sendSourceWebmail                = "webmail"
	sendSourceSubmission             = "submission"
	sendSourceOpenAPI                = "open_api"
	sendSourceForwarding             = "forwarding"
	sendSourceRuleForwarding         = "rule_forwarding"
	sendSourceForwardingVerification = "forwarding_verification"

	sendQueueStaleAfter  = 15 * time.Minute
	sendQueueConcurrency = 4

	sendQueueDeliveredMarkerDir = "send_queue_delivered"
)

type sendQueueInput struct {
	UserID        string
	MailboxID     string
	SentMessageID string
	MessageID     string
	Source        string
	MailFrom      string
	HeaderFrom    string
	Recipients    []string
	MIMEBytes     []byte
	Now           time.Time
}

type sendQueueItem struct {
	ID            string
	UserID        string
	MailboxID     string
	SentMessageID string
	MessageID     string
	Source        string
	MailFrom      string
	HeaderFrom    string
	Recipients    []string
	MIMEBytes     []byte
	AttemptCount  int
	MaxAttempts   int
}

func (a *App) enqueueSend(ctx context.Context, in sendQueueInput) (string, error) {
	if strings.TrimSpace(a.cfg.SMTPHost) == "" {
		return "", nil
	}
	now := in.Now.UTC()
	if now.IsZero() {
		now = a.now().UTC()
	}
	id := newID("snd")
	messageID := strings.TrimSpace(in.MessageID)
	mimeBase64 := base64.StdEncoding.EncodeToString(in.MIMEBytes)
	recipientsJSON := jsonEncode(dedupeEmails(in.Recipients))
	_, err := a.db.ExecContext(ctx, `INSERT OR IGNORE INTO send_queue(id,user_id,mailbox_id,sent_message_id,message_id,source,mail_from,header_from,recipients_json,mime_base64,status,next_attempt_at,created_at,updated_at)
		VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?)`,
		id, in.UserID, in.MailboxID, in.SentMessageID, messageID, in.Source, normalizeEmail(in.MailFrom), normalizeEmail(in.HeaderFrom), recipientsJSON, mimeBase64, sendQueueStatusQueued, now.Format(time.RFC3339Nano), now.Format(time.RFC3339Nano), now.Format(time.RFC3339Nano))
	if err != nil {
		return "", err
	}
	if strings.TrimSpace(messageID) != "" {
		var existingID, status string
		var attemptCount, maxAttempts int
		if err := a.db.QueryRowContext(ctx, `SELECT id,status,attempt_count,max_attempts FROM send_queue WHERE mailbox_id=? AND source=? AND message_id=?`, in.MailboxID, in.Source, messageID).Scan(&existingID, &status, &attemptCount, &maxAttempts); err != nil {
			return "", err
		}
		if existingID != id {
			if status == sendQueueStatusDelivered || status == sendQueueStatusCanceled || (status == sendQueueStatusFailed && attemptCount >= maxAttempts) {
				_, err := a.db.ExecContext(ctx, `UPDATE send_queue SET user_id=?,sent_message_id=?,mail_from=?,header_from=?,recipients_json=?,mime_base64=?,status=?,attempt_count=0,next_attempt_at=?,last_error='',updated_at=?,delivered_at=NULL WHERE id=? AND status=?`,
					in.UserID, in.SentMessageID, normalizeEmail(in.MailFrom), normalizeEmail(in.HeaderFrom), recipientsJSON, mimeBase64, sendQueueStatusQueued, now.Format(time.RFC3339Nano), now.Format(time.RFC3339Nano), existingID, status)
				if err != nil {
					return "", err
				}
				a.deleteSendQueueDeliveredMarker(existingID)
				a.recordSendAudit(ctx, sendAuditQueued, sendQueueStatusQueued, sendAuditInput{
					QueueID:       existingID,
					UserID:        in.UserID,
					MailboxID:     in.MailboxID,
					SentMessageID: in.SentMessageID,
					Source:        in.Source,
					MailFrom:      in.MailFrom,
					HeaderFrom:    in.HeaderFrom,
					Recipients:    in.Recipients,
				})
			}
			return existingID, nil
		}
	}
	a.recordSendAudit(ctx, sendAuditQueued, sendQueueStatusQueued, sendAuditInput{
		QueueID:       id,
		UserID:        in.UserID,
		MailboxID:     in.MailboxID,
		SentMessageID: in.SentMessageID,
		Source:        in.Source,
		MailFrom:      in.MailFrom,
		HeaderFrom:    in.HeaderFrom,
		Recipients:    in.Recipients,
	})
	return id, nil
}

func (a *App) sendQueueWorker(ctx context.Context) {
	a.log.Info("send queue worker started")
	ticker := time.NewTicker(10 * time.Second)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			a.log.Info("send queue worker stopped")
			return
		default:
		}
		if err := a.processDueSendQueue(ctx); err != nil {
			a.log.Warn("send queue worker failed", "error", err)
		}
		select {
		case <-ctx.Done():
			a.log.Info("send queue worker stopped")
			return
		case <-ticker.C:
		}
	}
}

func (a *App) processDueSendQueue(ctx context.Context) error {
	if strings.TrimSpace(a.cfg.SMTPHost) == "" {
		return nil
	}
	if err := a.recoverStaleSendQueueItems(ctx); err != nil {
		return err
	}
	rows, err := a.db.QueryContext(ctx, `SELECT id FROM send_queue WHERE (status=? OR (status=? AND attempt_count<max_attempts)) AND next_attempt_at<=? ORDER BY next_attempt_at, created_at LIMIT 20`, sendQueueStatusQueued, sendQueueStatusFailed, a.now().UTC().Format(time.RFC3339Nano))
	if err != nil {
		return err
	}
	defer rows.Close()
	var ids []string
	for rows.Next() {
		var id string
		if err := rows.Scan(&id); err != nil {
			return err
		}
		ids = append(ids, id)
	}
	if err := rows.Err(); err != nil {
		return err
	}
	sem := make(chan struct{}, sendQueueConcurrency)
	done := make(chan struct{}, len(ids))
	for _, id := range ids {
		select {
		case <-ctx.Done():
			return ctx.Err()
		case sem <- struct{}{}:
		}
		go func(id string) {
			defer func() {
				<-sem
				done <- struct{}{}
			}()
			a.processSendQueueItem(ctx, id)
		}(id)
	}
	for range ids {
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-done:
		}
	}
	return nil
}

func (a *App) recoverStaleSendQueueItems(ctx context.Context) error {
	cutoff := a.now().UTC().Add(-sendQueueStaleAfter).Format(time.RFC3339Nano)
	rows, err := a.db.QueryContext(ctx, `SELECT id,user_id,mailbox_id,sent_message_id,message_id,source,mail_from,header_from,recipients_json,mime_base64,attempt_count,max_attempts FROM send_queue WHERE status=? AND updated_at<=? AND attempt_count<max_attempts LIMIT 20`, sendQueueStatusSending, cutoff)
	if err != nil {
		return err
	}
	defer rows.Close()
	var items []sendQueueItem
	for rows.Next() {
		var item sendQueueItem
		var recipientsJSON, mimeBase64 string
		if err := rows.Scan(&item.ID, &item.UserID, &item.MailboxID, &item.SentMessageID, &item.MessageID, &item.Source, &item.MailFrom, &item.HeaderFrom, &recipientsJSON, &mimeBase64, &item.AttemptCount, &item.MaxAttempts); err != nil {
			return err
		}
		item.Recipients = jsonDecodeSlice(recipientsJSON)
		delivered, err := a.hasSendQueueDeliveredMarker(item.ID)
		if err != nil {
			return err
		}
		if delivered {
			items = append(items, item)
			continue
		}
		mimeBytes, err := base64.StdEncoding.DecodeString(mimeBase64)
		if err != nil {
			return err
		}
		item.MIMEBytes = mimeBytes
		items = append(items, item)
	}
	if err := rows.Err(); err != nil {
		return err
	}
	now := a.now().UTC().Format(time.RFC3339Nano)
	for _, item := range items {
		delivered, err := a.hasSendQueueDeliveredMarker(item.ID)
		if err != nil {
			return err
		}
		if delivered {
			if err := a.markSendQueueDelivered(ctx, item); err != nil {
				return err
			}
			continue
		}
		res, err := a.db.ExecContext(ctx, `UPDATE send_queue SET status=?,next_attempt_at=?,last_error=?,updated_at=? WHERE id=? AND status=?`, sendQueueStatusQueued, now, "send attempt interrupted", now, item.ID, sendQueueStatusSending)
		if err != nil {
			return err
		}
		if n, _ := res.RowsAffected(); n > 0 {
			a.recordSendAudit(ctx, sendAuditRetry, sendQueueStatusQueued, sendAuditInputFromQueue(item, "send attempt interrupted"))
		}
	}
	return nil
}

func (a *App) processSendQueueItem(ctx context.Context, id string) {
	item, err := a.claimSendQueueItem(ctx, id)
	if err != nil {
		if !errors.Is(err, sql.ErrNoRows) {
			a.log.Warn("failed to claim send queue item", "id", id, "error", err)
		}
		return
	}
	if err := a.sendSMTP(item.MailFrom, item.Recipients, item.MIMEBytes); err != nil {
		a.markSendQueueFailed(ctx, item, err)
		return
	}
	if err := a.writeSendQueueDeliveredMarker(item.ID); err != nil {
		a.log.Warn("failed to persist send queue delivered marker", "id", item.ID, "error", err)
	}
	if err := a.markSendQueueDelivered(ctx, item); err != nil {
		a.log.Warn("failed to mark send queue delivered", "id", item.ID, "error", err)
		return
	}
}

func (a *App) markSendQueueDelivered(ctx context.Context, item sendQueueItem) error {
	now := a.now().UTC().Format(time.RFC3339Nano)
	if _, err := a.db.ExecContext(ctx, `UPDATE send_queue SET status=?,delivered_at=?,updated_at=?,last_error='',mime_base64='' WHERE id=?`, sendQueueStatusDelivered, now, now, item.ID); err != nil {
		return err
	}
	a.deleteSendQueueDeliveredMarker(item.ID)
	a.recordSendAudit(ctx, sendAuditDelivered, sendQueueStatusDelivered, sendAuditInputFromQueue(item, ""))
	return nil
}

func (a *App) claimSendQueueItem(ctx context.Context, id string) (sendQueueItem, error) {
	now := a.now().UTC().Format(time.RFC3339Nano)
	res, err := a.db.ExecContext(ctx, `UPDATE send_queue SET status=?,attempt_count=attempt_count+1,updated_at=? WHERE id=? AND (status=? OR (status=? AND attempt_count<max_attempts)) AND next_attempt_at<=?`, sendQueueStatusSending, now, id, sendQueueStatusQueued, sendQueueStatusFailed, now)
	if err != nil {
		return sendQueueItem{}, err
	}
	if n, _ := res.RowsAffected(); n == 0 {
		return sendQueueItem{}, sql.ErrNoRows
	}
	var item sendQueueItem
	var recipientsJSON, mimeBase64 string
	row := a.db.QueryRowContext(ctx, `SELECT id,user_id,mailbox_id,sent_message_id,message_id,source,mail_from,header_from,recipients_json,mime_base64,attempt_count,max_attempts FROM send_queue WHERE id=?`, id)
	if err := row.Scan(&item.ID, &item.UserID, &item.MailboxID, &item.SentMessageID, &item.MessageID, &item.Source, &item.MailFrom, &item.HeaderFrom, &recipientsJSON, &mimeBase64, &item.AttemptCount, &item.MaxAttempts); err != nil {
		return sendQueueItem{}, err
	}
	item.Recipients = jsonDecodeSlice(recipientsJSON)
	mimeBytes, err := base64.StdEncoding.DecodeString(mimeBase64)
	if err != nil {
		return sendQueueItem{}, err
	}
	item.MIMEBytes = mimeBytes
	return item, nil
}

func (a *App) markSendQueueFailed(ctx context.Context, item sendQueueItem, sendErr error) {
	now := a.now().UTC()
	status := sendQueueStatusFailed
	nextAttempt := now.Add(sendRetryDelay(item.AttemptCount))
	if item.AttemptCount >= item.MaxAttempts {
		nextAttempt = now.Add(365 * 24 * time.Hour)
	}
	_, err := a.db.ExecContext(ctx, `UPDATE send_queue SET status=?,next_attempt_at=?,last_error=?,updated_at=? WHERE id=?`, status, nextAttempt.Format(time.RFC3339Nano), sendErr.Error(), now.Format(time.RFC3339Nano), item.ID)
	if err != nil {
		a.log.Warn("failed to mark send queue failed", "id", item.ID, "error", err)
	}
	event := sendAuditRetry
	if item.AttemptCount >= item.MaxAttempts {
		event = sendAuditFailed
	}
	a.recordSendAudit(ctx, event, status, sendAuditInputFromQueue(item, sendErr.Error()))
}

func sendRetryDelay(attempt int) time.Duration {
	if attempt < 1 {
		attempt = 1
	}
	delays := []time.Duration{30 * time.Second, 2 * time.Minute, 10 * time.Minute, time.Hour, 6 * time.Hour}
	if attempt > len(delays) {
		return delays[len(delays)-1]
	}
	return delays[attempt-1]
}

type sendAuditInput struct {
	QueueID       string
	UserID        string
	MailboxID     string
	SentMessageID string
	Source        string
	MailFrom      string
	HeaderFrom    string
	Recipients    []string
	Error         string
}

func sendAuditInputFromQueue(item sendQueueItem, errorText string) sendAuditInput {
	return sendAuditInput{
		QueueID:       item.ID,
		UserID:        item.UserID,
		MailboxID:     item.MailboxID,
		SentMessageID: item.SentMessageID,
		Source:        item.Source,
		MailFrom:      item.MailFrom,
		HeaderFrom:    item.HeaderFrom,
		Recipients:    item.Recipients,
		Error:         errorText,
	}
}

func (a *App) recordSendAudit(ctx context.Context, event, status string, in sendAuditInput) {
	source := strings.TrimSpace(in.Source)
	if source == "" {
		source = "unknown"
	}
	id := newID("audit")
	createdAt := a.now().UTC()
	item := SendAuditEvent{ID: id, QueueID: in.QueueID, MailboxID: in.MailboxID, SentMessageID: in.SentMessageID, Source: source, Event: event, Status: status, MailFrom: normalizeEmail(in.MailFrom), HeaderFrom: normalizeEmail(in.HeaderFrom), Recipients: dedupeEmails(in.Recipients), Error: in.Error, CreatedAt: createdAt}
	tx, err := a.db.BeginTx(ctx, nil)
	if err != nil {
		a.log.Warn("failed to start send audit transaction", "event", event, "error", err)
		return
	}
	defer tx.Rollback()
	if _, err := tx.ExecContext(ctx, `INSERT INTO send_audit_events(id,queue_id,user_id,mailbox_id,sent_message_id,source,event,status,mail_from,header_from,recipients_json,error,created_at)
		VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?)`, id, in.QueueID, in.UserID, in.MailboxID, in.SentMessageID, source, event, status, item.MailFrom, item.HeaderFrom, jsonEncode(item.Recipients), in.Error, createdAt.Format(time.RFC3339Nano)); err != nil {
		a.log.Warn("failed to record send audit", "event", event, "error", err)
		return
	}
	if err := a.enqueueStatusWebhook(ctx, tx, "audit:"+id, "send."+event, in.MailboxID, item); err != nil {
		a.log.Warn("failed to enqueue send status webhook", "event", event, "error", err)
		return
	}
	if err := tx.Commit(); err != nil {
		a.log.Warn("failed to commit send audit", "event", event, "error", err)
	}
}

func (a *App) sendQueueDeliveredMarkerPath(id string) string {
	safeID := filepath.Base(strings.TrimSpace(id))
	if safeID == "" || safeID == "." {
		safeID = "unknown"
	}
	return filepath.Join(a.cfg.DataDir, sendQueueDeliveredMarkerDir, safeID+".marker")
}

func (a *App) writeSendQueueDeliveredMarker(id string) error {
	path := a.sendQueueDeliveredMarkerPath(id)
	dir := filepath.Dir(path)
	if err := os.MkdirAll(dir, 0o700); err != nil {
		return err
	}
	tmp := filepath.Join(dir, filepath.Base(path)+"."+newID("tmp"))
	if err := os.WriteFile(tmp, []byte(a.now().UTC().Format(time.RFC3339Nano)), 0o600); err != nil {
		return err
	}
	if err := os.Rename(tmp, path); err != nil {
		_ = os.Remove(tmp)
		return err
	}
	return nil
}

func (a *App) hasSendQueueDeliveredMarker(id string) (bool, error) {
	_, err := os.Stat(a.sendQueueDeliveredMarkerPath(id))
	if err == nil {
		return true, nil
	}
	if errors.Is(err, os.ErrNotExist) {
		return false, nil
	}
	return false, err
}

func (a *App) deleteSendQueueDeliveredMarker(id string) {
	err := os.Remove(a.sendQueueDeliveredMarkerPath(id))
	if err != nil && !errors.Is(err, os.ErrNotExist) {
		a.log.Warn("failed to remove send queue delivered marker", "id", id, "error", err)
	}
}

func (a *App) authorizedSender(ctx context.Context, mb *Mailbox, from string) (string, string, error) {
	from = normalizeEmail(from)
	if from == "" {
		from = normalizeEmail(mb.Address)
	}
	if from == normalizeEmail(mb.Address) {
		return normalizeEmail(mb.Address), mb.DisplayName, nil
	}
	var displayName string
	var enabled int
	err := a.db.QueryRowContext(ctx, `SELECT display_name,enabled FROM send_as_grants WHERE mailbox_id=? AND address=?`, mb.ID, from).Scan(&displayName, &enabled)
	if err == nil {
		if enabled == 0 {
			return "", "", errSenderNotAuthorized
		}
		return from, strings.TrimSpace(displayName), nil
	}
	if !errors.Is(err, sql.ErrNoRows) {
		return "", "", err
	}
	var aliasDestination string
	err = a.db.QueryRowContext(ctx, `SELECT destination FROM aliases WHERE source=? AND enabled=1`, from).Scan(&aliasDestination)
	if err == nil {
		for _, destination := range strings.Split(aliasDestination, ",") {
			if normalizeEmail(destination) == normalizeEmail(mb.Address) {
				return from, mb.DisplayName, nil
			}
		}
	}
	return "", "", errSenderNotAuthorized
}