refactor(submission): 重构发件入库的去重流程。

- 将发件消息插入与附件保存改为事务内执行，避免去重键存在但消息缺失时产生脏数据。
- 抽出可注入的数据库执行器，便于在同一事务中复用消息与附件写入逻辑。
- 补充异常场景测试，覆盖“去重键存在但未命中消息”的失败行为。

apps/api/internal/app/app_test.go

@@ -12,6 +12,7 @@ import (
 	"encoding/base64"
 	"encoding/json"
 	"encoding/pem"
+	"errors"
 	"io"
 	"log/slog"
 	"math/big"
@@ -1179,6 +1180,46 @@ func TestSubmissionSentCopyDedupesByMessageID(t *testing.T) {
 	}
 }
 
+func TestInsertSentMessageOnceFailsWhenDedupeKeyHasNoMessage(t *testing.T) {
+	a := newTestApp(t)
+	_, mb := defaultAdminUserAndMailbox(t, a)
+	ctx := context.Background()
+	sentFolderID, err := a.ensureFolder(ctx, mb.ID, "Sent")
+	if err != nil {
+		t.Fatal(err)
+	}
+	messageID := "<orphan-sent-dedupe@example.test>"
+	if err := a.insertSentDedupeKey(ctx, mb.ID, sentFolderID, messageID); err != nil {
+		t.Fatal(err)
+	}
+	now := a.now().UTC()
+	sentID, inserted, err := a.insertSentMessageOnce(ctx, storedMessage{
+		MailboxID:  mb.ID,
+		MessageUID: newID("uid"),
+		MessageID:  messageID,
+		Subject:    "orphan dedupe",
+		From:       mb.Address,
+		To:         []string{"person@example.com"},
+		SentAt:     now,
+		ReceivedAt: now,
+		BodyText:   "body",
+		IsRead:     true,
+	}, nil)
+	if !errors.Is(err, errSentDedupeExists) {
+		t.Fatalf("insertSentMessageOnce error=%v, want errSentDedupeExists", err)
+	}
+	if sentID != "" || inserted {
+		t.Fatalf("sentID=%q inserted=%v, want empty false", sentID, inserted)
+	}
+	var count int
+	if err := a.db.QueryRow(`SELECT COUNT(1) FROM messages WHERE mailbox_id=? AND folder_id=? AND message_id=?`, mb.ID, sentFolderID, messageID).Scan(&count); err != nil {
+		t.Fatal(err)
+	}
+	if count != 0 {
+		t.Fatalf("orphan dedupe should not create sent message, count=%d", count)
+	}
+}
+
 func TestSubmissionRequeuesTerminalFailedDuplicateMessageID(t *testing.T) {
 	a := newTestApp(t)
 	a.cfg.SMTPHost = "127.0.0.1"

apps/api/internal/app/mail_handlers.go

@@ -1338,6 +1338,18 @@ func (a *App) messageByID(ctx context.Context, id string, includeBody bool) (*Ma
 }
 
 func (a *App) insertMessage(ctx context.Context, msg storedMessage, attachments []AttachmentInput) (string, error) {
+	return a.insertMessageWithDB(ctx, a.db, msg, attachments)
+}
+
+type dbExecutor interface {
+	ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error)
+}
+
+type dbQueryer interface {
+	QueryRowContext(ctx context.Context, query string, args ...any) *sql.Row
+}
+
+func (a *App) insertMessageWithDB(ctx context.Context, db dbExecutor, msg storedMessage, attachments []AttachmentInput) (string, error) {
 	id := newID("mail")
 	now := a.now().UTC().Format(time.RFC3339Nano)
 	hasAttachments := len(attachments) > 0
@@ -1355,13 +1367,14 @@ func (a *App) insertMessage(ctx context.Context, msg storedMessage, attachments
 		folderID = msg.FolderID
 	}
 	recipientAddr := normalizeEmail(msg.RecipientAddr)
-	_, err := a.db.ExecContext(ctx, `INSERT INTO messages(id,mailbox_id,folder_id,recipient_addr,message_uid,message_id,subject,from_addr,from_name,to_addrs,cc_addrs,bcc_addrs,sent_at,received_at,snippet,body_text,body_html,is_read,is_starred,has_attachments,size_bytes,raw_path,created_at,updated_at)
+	_, err := db.ExecContext(ctx, `INSERT INTO messages(id,mailbox_id,folder_id,recipient_addr,message_uid,message_id,subject,from_addr,from_name,to_addrs,cc_addrs,bcc_addrs,sent_at,received_at,snippet,body_text,body_html,is_read,is_starred,has_attachments,size_bytes,raw_path,created_at,updated_at)
 		VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`, id, mailboxID, folderID, recipientAddr, msg.MessageUID, msg.MessageID, msg.Subject, msg.From, msg.FromName, jsonEncode(msg.To), jsonEncode(msg.CC), jsonEncode(msg.BCC), msg.SentAt.Format(time.RFC3339Nano), msg.ReceivedAt.Format(time.RFC3339Nano), msg.Snippet, msg.BodyText, msg.BodyHTML, boolInt(msg.IsRead), boolInt(msg.IsStarred), boolInt(hasAttachments), size, msg.RawPath, now, now)
 	if err != nil {
 		return "", err
 	}
 	for _, att := range attachments {
-		if err := a.storeAttachment(ctx, id, att); err != nil {
+		if err := a.storeAttachmentWithDB(ctx, db, id, att); err != nil {
+			a.deleteMessageFiles(ctx, id)
 			return "", err
 		}
 	}
@@ -1369,6 +1382,10 @@ func (a *App) insertMessage(ctx context.Context, msg storedMessage, attachments
 }
 
 func (a *App) storeAttachment(ctx context.Context, messageID string, input AttachmentInput) error {
+	return a.storeAttachmentWithDB(ctx, a.db, messageID, input)
+}
+
+func (a *App) storeAttachmentWithDB(ctx context.Context, db dbExecutor, messageID string, input AttachmentInput) error {
 	filename := filepath.Base(strings.TrimSpace(input.Filename))
 	if filename == "." || filename == "" {
 		filename = "attachment.bin"
@@ -1390,7 +1407,7 @@ func (a *App) storeAttachment(ctx context.Context, messageID string, input Attac
 	if err := os.WriteFile(path, data, 0o600); err != nil {
 		return err
 	}
-	_, err = a.db.ExecContext(ctx, `INSERT INTO attachments(id,message_id,filename,content_type,size_bytes,storage_path,created_at) VALUES(?,?,?,?,?,?,?)`, id, messageID, filename, contentType, len(data), path, a.now().UTC().Format(time.RFC3339Nano))
+	_, err = db.ExecContext(ctx, `INSERT INTO attachments(id,message_id,filename,content_type,size_bytes,storage_path,created_at) VALUES(?,?,?,?,?,?,?)`, id, messageID, filename, contentType, len(data), path, a.now().UTC().Format(time.RFC3339Nano))
 	return err
 }
 

apps/api/internal/app/submission.go

@@ -322,46 +322,76 @@ func (a *App) insertSentMessageOnce(ctx context.Context, msg storedMessage, atta
 	if msg.MessageUID == "" {
 		msg.MessageUID = newID("uid")
 	}
+	tx, err := a.db.BeginTx(ctx, nil)
+	if err != nil {
+		return "", false, err
+	}
+	committed := false
+	messageIDForCleanup := ""
+	defer func() {
+		if !committed {
+			_ = tx.Rollback()
+			if messageIDForCleanup != "" {
+				a.deleteMessageFiles(ctx, messageIDForCleanup)
+			}
+		}
+	}()
 	if msg.MessageID != "" {
-		var existing string
-		err := a.db.QueryRowContext(ctx, `SELECT id FROM messages WHERE mailbox_id=? AND folder_id=? AND message_id=? AND message_id <> '' LIMIT 1`, msg.MailboxID, sentFolderID, msg.MessageID).Scan(&existing)
+		existing, err := sentMessageIDByMessageID(ctx, tx, msg.MailboxID, sentFolderID, msg.MessageID)
 		if err == nil {
-			if err := a.insertSentDedupeKey(ctx, msg.MailboxID, sentFolderID, msg.MessageID); err != nil && !errors.Is(err, errSentDedupeExists) {
+			if err := a.insertSentDedupeKeyWithDB(ctx, tx, msg.MailboxID, sentFolderID, msg.MessageID); err != nil && !errors.Is(err, errSentDedupeExists) {
 				return "", false, err
 			}
+			if err := tx.Commit(); err != nil {
+				return "", false, err
+			}
+			committed = true
 			return existing, false, nil
 		}
 		if err != nil && !errors.Is(err, sql.ErrNoRows) {
 			return "", false, err
 		}
-		if err := a.insertSentDedupeKey(ctx, msg.MailboxID, sentFolderID, msg.MessageID); err != nil {
+		if err := a.insertSentDedupeKeyWithDB(ctx, tx, msg.MailboxID, sentFolderID, msg.MessageID); err != nil {
 			if errors.Is(err, errSentDedupeExists) {
-				var existing string
-				if err := a.db.QueryRowContext(ctx, `SELECT id FROM messages WHERE mailbox_id=? AND folder_id=? AND message_id=? AND message_id <> '' LIMIT 1`, msg.MailboxID, sentFolderID, msg.MessageID).Scan(&existing); err == nil {
+				existing, qerr := sentMessageIDByMessageID(ctx, tx, msg.MailboxID, sentFolderID, msg.MessageID)
+				if qerr == nil {
+					if err := tx.Commit(); err != nil {
+						return "", false, err
+					}
+					committed = true
 					return existing, false, nil
 				}
-				return "", false, nil
+				if errors.Is(qerr, sql.ErrNoRows) {
+					return "", false, fmt.Errorf("sent dedupe key exists without sent message: %w", errSentDedupeExists)
+				}
+				return "", false, qerr
 			}
 			return "", false, err
 		}
 	}
-	id, err := a.insertMessage(ctx, msg, attachments)
+	id, err := a.insertMessageWithDB(ctx, tx, msg, attachments)
 	if err != nil {
-		if msg.MessageID != "" {
-			a.deleteSentDedupeKey(ctx, msg.MailboxID, sentFolderID, msg.MessageID)
-		}
 		return "", false, err
 	}
+	messageIDForCleanup = id
+	if err := tx.Commit(); err != nil {
+		return "", false, err
+	}
+	committed = true
 	return id, true, nil
 }
 
 var errSentDedupeExists = errors.New("sent message already exists")
 
 func (a *App) insertSentDedupeKey(ctx context.Context, mailboxID, folderID, messageID string) error {
+	return a.insertSentDedupeKeyWithDB(ctx, a.db, mailboxID, folderID, messageID)
+}
+
+func (a *App) insertSentDedupeKeyWithDB(ctx context.Context, db dbExecutor, mailboxID, folderID, messageID string) error {
 	if strings.TrimSpace(messageID) == "" {
 		return nil
 	}
-	res, err := a.db.ExecContext(ctx, `INSERT OR IGNORE INTO sent_message_dedupe_keys(mailbox_id,folder_id,message_id,created_at) VALUES(?,?,?,?)`, mailboxID, folderID, messageID, a.now().UTC().Format(time.RFC3339Nano))
+	res, err := db.ExecContext(ctx, `INSERT OR IGNORE INTO sent_message_dedupe_keys(mailbox_id,folder_id,message_id,created_at) VALUES(?,?,?,?)`, mailboxID, folderID, messageID, a.now().UTC().Format(time.RFC3339Nano))
 	if err != nil {
 		return err
 	}
@@ -371,6 +401,12 @@ func (a *App) insertSentDedupeKey(ctx context.Context, mailboxID, folderID, mess
 	return nil
 }
 
+func sentMessageIDByMessageID(ctx context.Context, db dbQueryer, mailboxID, folderID, messageID string) (string, error) {
+	var existing string
+	err := db.QueryRowContext(ctx, `SELECT id FROM messages WHERE mailbox_id=? AND folder_id=? AND message_id=? AND message_id <> '' LIMIT 1`, mailboxID, folderID, messageID).Scan(&existing)
+	return existing, err
+}
+
 func (a *App) deleteSentDedupeKey(ctx context.Context, mailboxID, folderID, messageID string) {
 	if strings.TrimSpace(messageID) == "" {
 		return