From d3bee62acdacf34af670bd8d3a9362d7e30d7eaf Mon Sep 17 00:00:00 2001 From: LanQin_ Date: Wed, 24 Jun 2026 13:26:53 +0800 Subject: [PATCH] =?UTF-8?q?refactor(submission):=20=E9=87=8D=E6=9E=84?= =?UTF-8?q?=E5=8F=91=E4=BB=B6=E5=85=A5=E5=BA=93=E7=9A=84=E5=8E=BB=E9=87=8D?= =?UTF-8?q?=E6=B5=81=E7=A8=8B=E3=80=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 将发件消息插入与附件保存改为事务内执行，避免去重键存在但消息缺失时产生脏数据。 - 抽出可注入的数据库执行器，便于在同一事务中复用消息与附件写入逻辑。 - 补充异常场景测试，覆盖“去重键存在但未命中消息”的失败行为。 --- apps/api/internal/app/app_test.go | 41 ++++++++++++++++++ apps/api/internal/app/mail_handlers.go | 23 ++++++++-- apps/api/internal/app/submission.go | 60 ++++++++++++++++++++------ 3 files changed, 109 insertions(+), 15 deletions(-) diff --git a/apps/api/internal/app/app_test.go b/apps/api/internal/app/app_test.go index 7e16843..78587c7 100644 --- a/apps/api/internal/app/app_test.go +++ b/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 := "" + 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" diff --git a/apps/api/internal/app/mail_handlers.go b/apps/api/internal/app/mail_handlers.go index f49b6a5..55ddcf4 100644 --- a/apps/api/internal/app/mail_handlers.go +++ b/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 } diff --git a/apps/api/internal/app/submission.go b/apps/api/internal/app/submission.go index d81b3db..27195a0 100644 --- a/apps/api/internal/app/submission.go +++ b/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