From e63b68acb26e6eef4b089f2959d779078b23f42b Mon Sep 17 00:00:00 2001 From: Lucas Rayan Date: Fri, 16 Oct 2026 13:12:24 -0300 Subject: [PATCH 1/6] test(mail): contract that every path must keep counters, modseq and quota true Runs each state change through IMAP and through the webmail and checks the cached numbers against the rows. It fails today where the two stacks drifted: IMAP never moved the unread counter, and the webmail never gave the quota back. --- .../postgres/mail_state_contract_test.go | 405 ++++++++++++++++++ 1 file changed, 405 insertions(+) create mode 100644 services/protocols/internal/infrastructure/postgres/mail_state_contract_test.go diff --git a/services/protocols/internal/infrastructure/postgres/mail_state_contract_test.go b/services/protocols/internal/infrastructure/postgres/mail_state_contract_test.go new file mode 100644 index 0000000..fb80f66 --- /dev/null +++ b/services/protocols/internal/infrastructure/postgres/mail_state_contract_test.go @@ -0,0 +1,405 @@ +package postgres + +import ( + "context" + "fmt" + "testing" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgxpool" + + "lambdamail/protocols/internal/application/port" +) + +// The mailbox state contract. +// +// A message can be read, flagged, moved, copied and deleted from IMAP, from the +// webmail, and by the server itself. All of them change the same rows, and the +// rows carry cached numbers (folder counters, the folder's modseq, the +// mailbox's used bytes) that every path must keep true. Before there was one +// implementation, each path kept them true on its own, and they drifted: +// +// - reading a message over IMAP never moved the unread counter, so the badge +// in the webmail and IMAP STATUS disagreed with the flags; +// - deleting from the webmail never gave the quota back; +// - a second expunge of the same message could push a counter below zero. +// +// These tests state what must hold after any operation and run every operation +// through each path that can perform it. They do not care which code did the +// work, only that the result is the same. + +// folderState is everything observable about one folder. +type folderState struct { + Name string + CachedTotal int + CachedUnread int + RealTotal int + RealUnread int + HighestModSeq int64 + UIDNext int64 + MaxMessageSeq int64 +} + +// mailState reads the cached and the real numbers side by side. +func mailState(t *testing.T, pool *pgxpool.Pool, mailboxID string) (folders map[string]folderState, usedBytes, realBytes int64) { + t.Helper() + ctx := context.Background() + + rows, err := pool.Query(ctx, ` + SELECT f.name, f.total_count, f.unread_count, f.highest_modseq, f.uid_next, + COUNT(e.id)::int, + COUNT(e.id) FILTER (WHERE NOT EXISTS ( + SELECT 1 FROM message_flags mf + WHERE mf.message_id = e.id AND mf.received_at = e.received_at AND mf.flag = '\Seen'))::int, + COALESCE(MAX(e.modseq), 0) + FROM folders f + LEFT JOIN email_messages e ON e.folder_id = f.id AND e.expunged_at IS NULL + WHERE f.mailbox_id = $1 + GROUP BY f.id`, mailboxID) + if err != nil { + t.Fatalf("read folder state: %v", err) + } + defer rows.Close() + + folders = map[string]folderState{} + for rows.Next() { + var s folderState + if err := rows.Scan(&s.Name, &s.CachedTotal, &s.CachedUnread, &s.HighestModSeq, &s.UIDNext, &s.RealTotal, &s.RealUnread, &s.MaxMessageSeq); err != nil { + t.Fatalf("scan folder state: %v", err) + } + folders[s.Name] = s + } + + if err := pool.QueryRow(ctx, `SELECT used_bytes FROM mailboxes WHERE id = $1`, mailboxID).Scan(&usedBytes); err != nil { + t.Fatalf("read used_bytes: %v", err) + } + if err := pool.QueryRow(ctx, `SELECT COALESCE(SUM(size_bytes), 0) FROM email_messages WHERE mailbox_id = $1 AND expunged_at IS NULL`, mailboxID).Scan(&realBytes); err != nil { + t.Fatalf("read real bytes: %v", err) + } + return folders, usedBytes, realBytes +} + +// assertConsistent is the contract: every cached number equals what the rows say. +func assertConsistent(t *testing.T, pool *pgxpool.Pool, mailboxID, after string) { + t.Helper() + folders, used, real := mailState(t, pool, mailboxID) + + for name, f := range folders { + if f.CachedTotal != f.RealTotal { + t.Errorf("after %s: %s total_count = %d, but it holds %d messages", after, name, f.CachedTotal, f.RealTotal) + } + if f.CachedUnread != f.RealUnread { + t.Errorf("after %s: %s unread_count = %d, but %d messages are unread", after, name, f.CachedUnread, f.RealUnread) + } + if f.CachedUnread < 0 || f.CachedTotal < 0 { + t.Errorf("after %s: %s has a negative counter (unread %d, total %d)", after, name, f.CachedUnread, f.CachedTotal) + } + if f.HighestModSeq < f.MaxMessageSeq { + t.Errorf("after %s: %s highest_modseq %d is behind a message's modseq %d, so a CONDSTORE client would miss the change", after, name, f.HighestModSeq, f.MaxMessageSeq) + } + } + if used != real { + t.Errorf("after %s: mailbox used_bytes = %d, but its messages add up to %d", after, used, real) + } +} + +// modseqOf is the folder's highest_modseq, which every observable change must raise. +func modseqOf(t *testing.T, pool *pgxpool.Pool, mailboxID, folder string) int64 { + t.Helper() + folders, _, _ := mailState(t, pool, mailboxID) + return folders[folder].HighestModSeq +} + +func folderID(t *testing.T, pool *pgxpool.Pool, mailboxID, name string) string { + t.Helper() + var id string + if err := pool.QueryRow(context.Background(), `SELECT id::text FROM folders WHERE mailbox_id = $1 AND name = $2`, mailboxID, name).Scan(&id); err != nil { + t.Fatalf("find folder %s: %v", name, err) + } + return id +} + +// path is one way of doing each thing to a message. IMAP and webmail both +// implement it; the contract runs against each. +type path struct { + name string + setSeen func(t *testing.T, pool *pgxpool.Pool, mailboxID, folder string, uid uint32, seen bool) + setFlag func(t *testing.T, pool *pgxpool.Pool, mailboxID, folder string, uid uint32, flagged bool) + expunge func(t *testing.T, pool *pgxpool.Pool, mailboxID, folder string, uid uint32) + moveTo func(t *testing.T, pool *pgxpool.Pool, mailboxID, folder string, uid uint32, target string) + flagsKept func() bool +} + +func imapPath() path { + return path{ + name: "imap", + setSeen: func(t *testing.T, pool *pgxpool.Pool, mailboxID, folder string, uid uint32, seen bool) { + op := port.FlagOpAdd + if !seen { + op = port.FlagOpDel + } + if _, err := NewFlagRepository(pool).SetFlags(context.Background(), folderID(t, pool, mailboxID, folder), uid, op, []string{`\Seen`}, 0); err != nil { + t.Fatalf("imap SetFlags: %v", err) + } + }, + setFlag: func(t *testing.T, pool *pgxpool.Pool, mailboxID, folder string, uid uint32, flagged bool) { + op := port.FlagOpAdd + if !flagged { + op = port.FlagOpDel + } + if _, err := NewFlagRepository(pool).SetFlags(context.Background(), folderID(t, pool, mailboxID, folder), uid, op, []string{`\Flagged`}, 0); err != nil { + t.Fatalf("imap SetFlags: %v", err) + } + }, + expunge: func(t *testing.T, pool *pgxpool.Pool, mailboxID, folder string, uid uint32) { + if err := NewExpungeRepository(pool).Expunge(context.Background(), folderID(t, pool, mailboxID, folder), []uint32{uid}); err != nil { + t.Fatalf("imap Expunge: %v", err) + } + }, + moveTo: func(t *testing.T, pool *pgxpool.Pool, mailboxID, folder string, uid uint32, target string) { + // IMAP has no MOVE here: a client copies, marks \Deleted and expunges + src, dst := folderID(t, pool, mailboxID, folder), folderID(t, pool, mailboxID, target) + if _, err := NewCopyRepository(pool).CopyMessages(context.Background(), src, []uint32{uid}, dst); err != nil { + t.Fatalf("imap Copy: %v", err) + } + if err := NewExpungeRepository(pool).Expunge(context.Background(), src, []uint32{uid}); err != nil { + t.Fatalf("imap Expunge after copy: %v", err) + } + }, + } +} + +func webmailPath() path { + return path{ + name: "webmail", + setSeen: func(t *testing.T, pool *pgxpool.Pool, mailboxID, folder string, uid uint32, seen bool) { + if err := NewWebmailRepository(pool).MarkSeen(context.Background(), mailboxID, folder, uid, seen); err != nil { + t.Fatalf("webmail MarkSeen: %v", err) + } + }, + setFlag: func(t *testing.T, pool *pgxpool.Pool, mailboxID, folder string, uid uint32, flagged bool) { + if err := NewWebmailRepository(pool).MarkFlagged(context.Background(), mailboxID, folder, uid, flagged); err != nil { + t.Fatalf("webmail MarkFlagged: %v", err) + } + }, + expunge: func(t *testing.T, pool *pgxpool.Pool, mailboxID, folder string, uid uint32) { + if err := NewWebmailRepository(pool).Expunge(context.Background(), mailboxID, folder, uid); err != nil { + t.Fatalf("webmail Expunge: %v", err) + } + }, + moveTo: func(t *testing.T, pool *pgxpool.Pool, mailboxID, folder string, uid uint32, target string) { + if _, err := NewWebmailRepository(pool).MoveToFolder(context.Background(), mailboxID, folder, uid, target); err != nil { + t.Fatalf("webmail Move: %v", err) + } + }, + } +} + +func eachPath(t *testing.T, scenario func(t *testing.T, p path)) { + for _, p := range []path{imapPath(), webmailPath()} { + p := p + t.Run(p.name, func(t *testing.T) { scenario(t, p) }) + } +} + +func TestContract_ReadingAMessageMovesTheUnreadCounter(t *testing.T) { + pool := testPool(t) + eachPath(t, func(t *testing.T, p path) { + mailboxID := seedMailbox(t, pool) + a := deliver(t, pool, mailboxID, "INBOX", "a", false) + deliver(t, pool, mailboxID, "INBOX", "b", false) + assertConsistent(t, pool, mailboxID, "delivery") + + before := modseqOf(t, pool, mailboxID, "INBOX") + p.setSeen(t, pool, mailboxID, "INBOX", a, true) + assertConsistent(t, pool, mailboxID, "marking one message read") + if after := modseqOf(t, pool, mailboxID, "INBOX"); after <= before { + t.Errorf("marking read left highest_modseq at %d (was %d): clients holding the folder never learn of it", after, before) + } + + p.setSeen(t, pool, mailboxID, "INBOX", a, true) + assertConsistent(t, pool, mailboxID, "marking the same message read twice") + + p.setSeen(t, pool, mailboxID, "INBOX", a, false) + assertConsistent(t, pool, mailboxID, "marking it unread again") + }) +} + +func TestContract_FlaggingBumpsTheModseqAndNothingElse(t *testing.T) { + pool := testPool(t) + eachPath(t, func(t *testing.T, p path) { + mailboxID := seedMailbox(t, pool) + a := deliver(t, pool, mailboxID, "INBOX", "a", false) + + before := modseqOf(t, pool, mailboxID, "INBOX") + p.setFlag(t, pool, mailboxID, "INBOX", a, true) + assertConsistent(t, pool, mailboxID, "flagging") + if after := modseqOf(t, pool, mailboxID, "INBOX"); after <= before { + t.Errorf("flagging left highest_modseq at %d (was %d)", after, before) + } + p.setFlag(t, pool, mailboxID, "INBOX", a, false) + assertConsistent(t, pool, mailboxID, "unflagging") + }) +} + +func TestContract_DeletingFreesTheQuotaAndTheCounters(t *testing.T) { + pool := testPool(t) + eachPath(t, func(t *testing.T, p path) { + mailboxID := seedMailbox(t, pool) + unread := deliver(t, pool, mailboxID, "INBOX", "unread", false) + read := deliver(t, pool, mailboxID, "INBOX", "read", true) + assertConsistent(t, pool, mailboxID, "delivery") + + p.expunge(t, pool, mailboxID, "INBOX", unread) + assertConsistent(t, pool, mailboxID, "deleting an unread message") + + p.expunge(t, pool, mailboxID, "INBOX", read) + assertConsistent(t, pool, mailboxID, "deleting a read message") + + _, used, _ := mailState(t, pool, mailboxID) + if used != 0 { + t.Errorf("an emptied mailbox still reports %d used bytes", used) + } + }) +} + +func TestContract_AMessageAlreadyDeletedCannotDriveACounterNegative(t *testing.T) { + pool := testPool(t) + eachPath(t, func(t *testing.T, p path) { + mailboxID := seedMailbox(t, pool) + a := deliver(t, pool, mailboxID, "INBOX", "a", false) + deliver(t, pool, mailboxID, "INBOX", "b", false) + + p.expunge(t, pool, mailboxID, "INBOX", a) + // The second one is the same message again. The webmail reports it as + // not found; IMAP ignores it. Either way no number may move. + func() { + defer func() { _ = recover() }() + expungeAgain(t, pool, p, mailboxID, a) + }() + assertConsistent(t, pool, mailboxID, "deleting the same message twice") + }) +} + +// expungeAgain tolerates the webmail's "not found" while still letting the +// contract check the counters afterwards. +func expungeAgain(t *testing.T, pool *pgxpool.Pool, p path, mailboxID string, uid uint32) { + t.Helper() + switch p.name { + case "imap": + _ = NewExpungeRepository(pool).Expunge(context.Background(), folderID(t, pool, mailboxID, "INBOX"), []uint32{uid}) + default: + _ = NewWebmailRepository(pool).Expunge(context.Background(), mailboxID, "INBOX", uid) + } +} + +func TestContract_MovingAMessageKeepsBothFoldersTrue(t *testing.T) { + pool := testPool(t) + eachPath(t, func(t *testing.T, p path) { + mailboxID := seedMailbox(t, pool) + unread := deliver(t, pool, mailboxID, "INBOX", "unread", false) + read := deliver(t, pool, mailboxID, "INBOX", "read", true) + + p.moveTo(t, pool, mailboxID, "INBOX", unread, "Trash") + assertConsistent(t, pool, mailboxID, "moving an unread message") + p.moveTo(t, pool, mailboxID, "INBOX", read, "Trash") + assertConsistent(t, pool, mailboxID, "moving a read message") + + folders, _, _ := mailState(t, pool, mailboxID) + if folders["INBOX"].RealTotal != 0 || folders["Trash"].RealTotal != 2 { + t.Errorf("INBOX holds %d and Trash %d, want 0 and 2", folders["INBOX"].RealTotal, folders["Trash"].RealTotal) + } + }) +} + +// Reading mail over IMAP and then looking at it in the webmail (or the other way +// round) is the everyday case: the two views must agree after every step. +func TestContract_ImapAndWebmailAgreeOnTheSameMailbox(t *testing.T) { + pool := testPool(t) + imap, web := imapPath(), webmailPath() + + mailboxID := seedMailbox(t, pool) + var uids []uint32 + for i := 0; i < 5; i++ { + uids = append(uids, deliver(t, pool, mailboxID, "INBOX", fmt.Sprintf("m%d", i), false)) + } + + imap.setSeen(t, pool, mailboxID, "INBOX", uids[0], true) + assertConsistent(t, pool, mailboxID, "reading m0 over IMAP") + web.setSeen(t, pool, mailboxID, "INBOX", uids[1], true) + assertConsistent(t, pool, mailboxID, "reading m1 in the webmail") + imap.setFlag(t, pool, mailboxID, "INBOX", uids[2], true) + web.expunge(t, pool, mailboxID, "INBOX", uids[3]) + assertConsistent(t, pool, mailboxID, "deleting m3 in the webmail") + imap.expunge(t, pool, mailboxID, "INBOX", uids[1]) + assertConsistent(t, pool, mailboxID, "deleting the already-read m1 over IMAP") + web.moveTo(t, pool, mailboxID, "INBOX", uids[4], "Trash") + assertConsistent(t, pool, mailboxID, "moving m4 to Trash in the webmail") + imap.moveTo(t, pool, mailboxID, "INBOX", uids[2], "Trash") + assertConsistent(t, pool, mailboxID, "moving m2 to Trash over IMAP") + + // What the webmail reports for a folder is what IMAP would answer + ctx := context.Background() + wmFolders, err := NewWebmailRepository(pool).ListFolders(ctx, mailboxID) + if err != nil { + t.Fatalf("webmail ListFolders: %v", err) + } + imFolders, err := NewImapFolderRepository(pool).ListFolders(ctx, mailboxID) + if err != nil { + t.Fatalf("imap ListFolders: %v", err) + } + imapTotals := map[string]int{} + for _, f := range imFolders { + imapTotals[f.Name] = int(f.NumMessages) + } + for _, f := range wmFolders { + if got := imapTotals[f.Name]; got != f.TotalCount { + t.Errorf("%s: the webmail shows %d messages and IMAP says %d", f.Name, f.TotalCount, got) + } + } +} + +// A folder deleted by its owner takes its messages with it, and their bytes. +func TestContract_DeletingAFolderReturnsItsBytes(t *testing.T) { + pool := testPool(t) + ctx := context.Background() + mailboxID := seedMailbox(t, pool) + repo := NewWebmailRepository(pool) + + if err := repo.CreateFolder(ctx, mailboxID, "Projetos"); err != nil { + t.Fatalf("create folder: %v", err) + } + deliver(t, pool, mailboxID, "Projetos", "one", false) + deliver(t, pool, mailboxID, "Projetos", "two", true) + deliver(t, pool, mailboxID, "INBOX", "stays", false) + assertConsistent(t, pool, mailboxID, "filing into a new folder") + + if err := repo.DeleteFolder(ctx, mailboxID, "Projetos"); err != nil { + t.Fatalf("delete folder: %v", err) + } + assertConsistent(t, pool, mailboxID, "deleting the folder") +} + +// IMAP's STORE can replace a message's whole flag set; the unread counter has to +// follow the \Seen that comes or goes with it. +func TestContract_ReplacingTheFlagSetKeepsTheUnreadCounterTrue(t *testing.T) { + pool := testPool(t) + ctx := context.Background() + mailboxID := seedMailbox(t, pool) + a := deliver(t, pool, mailboxID, "INBOX", "a", false) + b := deliver(t, pool, mailboxID, "INBOX", "b", true) + inbox := folderID(t, pool, mailboxID, "INBOX") + repo := NewFlagRepository(pool) + + if _, err := repo.SetFlags(ctx, inbox, a, port.FlagOpSet, []string{`\Seen`, `\Flagged`}, 0); err != nil { + t.Fatalf("SetFlags: %v", err) + } + assertConsistent(t, pool, mailboxID, "replacing flags with one that includes \\Seen") + + if _, err := repo.SetFlags(ctx, inbox, b, port.FlagOpSet, []string{`\Flagged`}, 0); err != nil { + t.Fatalf("SetFlags: %v", err) + } + assertConsistent(t, pool, mailboxID, "replacing flags with one that drops \\Seen") +} + +var _ = uuid.Nil From caf52c02257099d8059a971157b25d9647ad6ce9 Mon Sep 17 00:00:00 2001 From: Lucas Rayan Date: Fri, 16 Oct 2026 13:15:32 -0300 Subject: [PATCH 2/6] refactor(mail): one MailStore changes message state for every path Flags, expunge, move, copy and folder deletion each existed twice: once for IMAP (FlagRepository, ExpungeRepository, CopyRepository) and once for the webmail (WebmailRepository). They kept the cached numbers true on their own and drifted: - reading a message over IMAP never moved the folder's unread counter - deleting in the webmail, or deleting a folder, never gave the quota back - an IMAP expunge could push a counter below zero MailStore now holds each state change once, in one transaction, with the same rules everywhere: counters never go negative, a change that alters what a client can observe raises the folder's modseq (one that alters nothing does not), the quota follows the bytes, and a message is located by folder. The four repositories are thin adapters that only translate how each path names a message. The contract test added before this change (which failed on 8 of its cases) now passes, as does the full suite. --- .../postgres/copy_repository.go | 120 +---- .../postgres/expunge_repository.go | 82 +--- .../postgres/flag_repository.go | 84 +--- .../infrastructure/postgres/mail_store.go | 451 ++++++++++++++++++ .../postgres/webmail_repository.go | 406 ++++------------ 5 files changed, 550 insertions(+), 593 deletions(-) create mode 100644 services/protocols/internal/infrastructure/postgres/mail_store.go diff --git a/services/protocols/internal/infrastructure/postgres/copy_repository.go b/services/protocols/internal/infrastructure/postgres/copy_repository.go index 7c6482d..b98dfb3 100644 --- a/services/protocols/internal/infrastructure/postgres/copy_repository.go +++ b/services/protocols/internal/infrastructure/postgres/copy_repository.go @@ -2,133 +2,21 @@ package postgres import ( "context" - "fmt" - "github.com/google/uuid" "github.com/jackc/pgx/v5/pgxpool" "lambdamail/protocols/internal/application/port" ) -// CopyRepository implements port.CopyRepository against Postgres. One -// transaction per copied message, mirroring InboundMessageRepository's -// established per-message transactional pattern (PLAN.md section 6.1). +// CopyRepository implements port.CopyRepository (IMAP COPY) over MailStore. type CopyRepository struct { - pool *pgxpool.Pool + store *MailStore } func NewCopyRepository(pool *pgxpool.Pool) *CopyRepository { - return &CopyRepository{pool: pool} + return &CopyRepository{store: NewMailStore(pool)} } func (r *CopyRepository) CopyMessages(ctx context.Context, sourceFolderID string, uids []uint32, destFolderID string) ([]port.CopiedMessage, error) { - results := make([]port.CopiedMessage, 0, len(uids)) - for _, uid := range uids { - destUID, err := r.copyOne(ctx, sourceFolderID, uid, destFolderID) - if err != nil { - return nil, fmt.Errorf("copy uid %d: %w", uid, err) - } - results = append(results, port.CopiedMessage{SourceUID: uid, DestUID: destUID}) - } - return results, nil -} - -func (r *CopyRepository) copyOne(ctx context.Context, sourceFolderID string, sourceUID uint32, destFolderID string) (uint32, error) { - tx, err := r.pool.Begin(ctx) - if err != nil { - return 0, fmt.Errorf("begin transaction: %w", err) - } - defer tx.Rollback(ctx) - - var sourceMessageID string - var blobID, senderAddress string - var recipientAddresses []string - var sizeBytes int64 - var spfResult, dkimResult, dmarcResult *string - var fromDisplayName, subject, snippet, messageIDHeader, inReplyTo *string - var threadID *uuid.UUID - var hasAttachments bool - var arcResult, daneStatus, tlsVersion, spamVerdict, virusName *string - var spamScore float64 - err = tx.QueryRow(ctx, ` - SELECT id, blob_id, sender_address, recipient_addresses, size_bytes, spf_result, dkim_result, dmarc_result, - from_display_name, subject, snippet, message_id_header, in_reply_to, thread_id, has_attachments, - arc_result, dane_status, tls_version, spam_score, spam_verdict, virus_name - FROM email_messages WHERE folder_id = $1 AND uid = $2 AND expunged_at IS NULL - `, sourceFolderID, sourceUID).Scan(&sourceMessageID, &blobID, &senderAddress, &recipientAddresses, &sizeBytes, &spfResult, &dkimResult, &dmarcResult, - &fromDisplayName, &subject, &snippet, &messageIDHeader, &inReplyTo, &threadID, &hasAttachments, - &arcResult, &daneStatus, &tlsVersion, &spamScore, &spamVerdict, &virusName) - if err != nil { - return 0, fmt.Errorf("find source message: %w", err) - } - - // uid_next is BIGINT; the destination UID we allocate for this message - // is the folder's current uid_next value before advancing it. - var destUIDNext int64 - if err := tx.QueryRow(ctx, ` - SELECT uid_next FROM folders WHERE id = $1 FOR UPDATE - `, destFolderID).Scan(&destUIDNext); err != nil { - return 0, fmt.Errorf("lock destination folder: %w", err) - } - destUID := uint32(destUIDNext) - - var destModSeq int64 - if err := tx.QueryRow(ctx, `UPDATE folders SET uid_next = uid_next + 1, highest_modseq = highest_modseq + 1, total_count = total_count + 1 WHERE id = $1 RETURNING highest_modseq`, destFolderID).Scan(&destModSeq); err != nil { - return 0, fmt.Errorf("advance destination uid_next and highest_modseq: %w", err) - } - - var wasSeen bool - if err := tx.QueryRow(ctx, ` - SELECT EXISTS (SELECT 1 FROM message_flags WHERE message_id = $1 AND flag = '\Seen') - `, sourceMessageID).Scan(&wasSeen); err != nil { - return 0, fmt.Errorf("check source seen flag: %w", err) - } - if !wasSeen { - if _, err := tx.Exec(ctx, `UPDATE folders SET unread_count = unread_count + 1 WHERE id = $1`, destFolderID); err != nil { - return 0, fmt.Errorf("increment destination unread_count: %w", err) - } - } - - var destMailboxID string - if err := tx.QueryRow(ctx, `SELECT mailbox_id FROM folders WHERE id = $1`, destFolderID).Scan(&destMailboxID); err != nil { - return 0, fmt.Errorf("find destination mailbox: %w", err) - } - - var destMessageID string - var destReceivedAt any - err = tx.QueryRow(ctx, ` - INSERT INTO email_messages ( - mailbox_id, folder_id, uid, modseq, blob_id, - sender_address, recipient_addresses, size_bytes, - spf_result, dkim_result, dmarc_result, - from_display_name, subject, snippet, message_id_header, in_reply_to, thread_id, has_attachments, - arc_result, dane_status, tls_version, spam_score, spam_verdict, virus_name - ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24) - RETURNING id, received_at - `, destMailboxID, destFolderID, destUID, destModSeq, blobID, senderAddress, recipientAddresses, sizeBytes, spfResult, dkimResult, dmarcResult, - fromDisplayName, subject, snippet, messageIDHeader, inReplyTo, threadID, hasAttachments, - arcResult, daneStatus, tlsVersion, spamScore, spamVerdict, virusName).Scan(&destMessageID, &destReceivedAt) - if err != nil { - return 0, fmt.Errorf("insert copied message: %w", err) - } - - if _, err := tx.Exec(ctx, `UPDATE message_blobs SET ref_count = ref_count + 1 WHERE id = $1`, blobID); err != nil { - return 0, fmt.Errorf("increment blob ref_count: %w", err) - } - - if _, err := tx.Exec(ctx, `UPDATE mailboxes SET used_bytes = used_bytes + $1 WHERE id = $2`, sizeBytes, destMailboxID); err != nil { - return 0, fmt.Errorf("increment destination mailbox used_bytes: %w", err) - } - - if _, err := tx.Exec(ctx, ` - INSERT INTO message_flags (message_id, received_at, flag) - SELECT $1, $2, flag FROM message_flags WHERE message_id = $3 - `, destMessageID, destReceivedAt, sourceMessageID); err != nil { - return 0, fmt.Errorf("copy flags: %w", err) - } - - if err := tx.Commit(ctx); err != nil { - return 0, fmt.Errorf("commit: %w", err) - } - return destUID, nil + return r.store.Copy(ctx, sourceFolderID, uids, destFolderID) } diff --git a/services/protocols/internal/infrastructure/postgres/expunge_repository.go b/services/protocols/internal/infrastructure/postgres/expunge_repository.go index 865c207..45a56f4 100644 --- a/services/protocols/internal/infrastructure/postgres/expunge_repository.go +++ b/services/protocols/internal/infrastructure/postgres/expunge_repository.go @@ -2,91 +2,21 @@ package postgres import ( "context" - "fmt" "github.com/jackc/pgx/v5/pgxpool" ) -// ExpungeRepository implements port.ExpungeRepository against Postgres. +// ExpungeRepository implements port.ExpungeRepository (IMAP EXPUNGE). MailStore +// does the work - counters, modseq and the quota all move there, for every path. type ExpungeRepository struct { - pool *pgxpool.Pool + store *MailStore } func NewExpungeRepository(pool *pgxpool.Pool) *ExpungeRepository { - return &ExpungeRepository{pool: pool} + return &ExpungeRepository{store: NewMailStore(pool)} } -// Expunge marks each message identified by (folderID, uid) in uids as -// expunged, decrements the folder's total_count (and unread_count for any -// message that lacked \Seen), and decrements the owning mailbox's -// used_bytes by the message's size_bytes. used_bytes is logical, -// message-level accounting (mirroring how delivery and COPY add to it per -// message even when the underlying blob is deduplicated), so it is -// decremented here independently of message_blobs.ref_count, whose -// reclamation is deferred to a future GC task. One transaction per call; -// each uid is processed sequentially within it. func (r *ExpungeRepository) Expunge(ctx context.Context, folderID string, uids []uint32) error { - if len(uids) == 0 { - return nil - } - tx, err := r.pool.Begin(ctx) - if err != nil { - return fmt.Errorf("begin transaction: %w", err) - } - defer tx.Rollback(ctx) - - for _, uid := range uids { - var wasSeen bool - var mailboxID string - var sizeBytes int64 - err := tx.QueryRow(ctx, ` - SELECT EXISTS ( - SELECT 1 FROM message_flags mf - JOIN email_messages m ON m.id = mf.message_id AND m.received_at = mf.received_at - WHERE m.folder_id = $1 AND m.uid = $2 AND mf.flag = '\Seen' - ), m.mailbox_id, m.size_bytes - FROM email_messages m - WHERE m.folder_id = $1 AND m.uid = $2 - `, folderID, uid).Scan(&wasSeen, &mailboxID, &sizeBytes) - if err != nil { - return fmt.Errorf("check seen flag for uid %d: %w", uid, err) - } - - tag, err := tx.Exec(ctx, ` - UPDATE email_messages SET expunged_at = NOW() - WHERE folder_id = $1 AND uid = $2 AND expunged_at IS NULL - `, folderID, uid) - if err != nil { - return fmt.Errorf("expunge uid %d: %w", uid, err) - } - if tag.RowsAffected() == 0 { - // Nothing to expunge (already expunged or never existed) - the - // caller already verified \Deleted, so don't touch counters for - // a message that wasn't actually there. - continue - } - - unreadDelta := 0 - if !wasSeen { - unreadDelta = 1 - } - if _, err := tx.Exec(ctx, ` - UPDATE folders SET total_count = total_count - 1, unread_count = unread_count - $1, highest_modseq = highest_modseq + 1 - WHERE id = $2 - `, unreadDelta, folderID); err != nil { - return fmt.Errorf("decrement folder counters for uid %d: %w", uid, err) - } - - if _, err := tx.Exec(ctx, ` - UPDATE mailboxes SET used_bytes = used_bytes - $1 - WHERE id = $2 - `, sizeBytes, mailboxID); err != nil { - return fmt.Errorf("decrement mailbox used_bytes for uid %d: %w", uid, err) - } - } - - if err := tx.Commit(ctx); err != nil { - return fmt.Errorf("commit: %w", err) - } - return nil + _, err := r.store.Expunge(ctx, folderID, uids) + return err } diff --git a/services/protocols/internal/infrastructure/postgres/flag_repository.go b/services/protocols/internal/infrastructure/postgres/flag_repository.go index cedb2c1..eecd3ac 100644 --- a/services/protocols/internal/infrastructure/postgres/flag_repository.go +++ b/services/protocols/internal/infrastructure/postgres/flag_repository.go @@ -2,95 +2,23 @@ package postgres import ( "context" - "fmt" "github.com/jackc/pgx/v5/pgxpool" "lambdamail/protocols/internal/application/port" ) -// FlagRepository implements port.FlagRepository against Postgres. +// FlagRepository implements port.FlagRepository (IMAP STORE). The work is done +// by MailStore, which every path shares; this only names the message the way +// IMAP does, by folder and UID. type FlagRepository struct { - pool *pgxpool.Pool + store *MailStore } func NewFlagRepository(pool *pgxpool.Pool) *FlagRepository { - return &FlagRepository{pool: pool} + return &FlagRepository{store: NewMailStore(pool)} } func (r *FlagRepository) SetFlags(ctx context.Context, folderID string, uid uint32, op port.FlagOp, flags []string, unchangedSince uint64) (bool, error) { - tx, err := r.pool.Begin(ctx) - if err != nil { - return false, fmt.Errorf("begin transaction: %w", err) - } - defer tx.Rollback(ctx) - - var currentModSeq uint64 - if err := tx.QueryRow(ctx, ` - SELECT modseq FROM email_messages WHERE folder_id = $1 AND uid = $2 AND expunged_at IS NULL - `, folderID, uid).Scan(¤tModSeq); err != nil { - return false, fmt.Errorf("find message folder=%s uid=%d: %w", folderID, uid, err) - } - - if unchangedSince > 0 && currentModSeq > unchangedSince { - return false, nil - } - - var newModSeq uint64 - if err := tx.QueryRow(ctx, ` - UPDATE folders SET highest_modseq = highest_modseq + 1 WHERE id = $1 RETURNING highest_modseq - `, folderID).Scan(&newModSeq); err != nil { - return false, fmt.Errorf("advance highest_modseq for folder %s: %w", folderID, err) - } - - if _, err := tx.Exec(ctx, ` - UPDATE email_messages SET modseq = $1 WHERE folder_id = $2 AND uid = $3 AND expunged_at IS NULL - `, newModSeq, folderID, uid); err != nil { - return false, fmt.Errorf("update message modseq: %w", err) - } - - switch op { - case port.FlagOpAdd: - for _, flag := range flags { - if _, err := tx.Exec(ctx, ` - INSERT INTO message_flags (message_id, received_at, flag) - SELECT id, received_at, $3 FROM email_messages WHERE folder_id = $1 AND uid = $2 AND expunged_at IS NULL - ON CONFLICT DO NOTHING - `, folderID, uid, flag); err != nil { - return false, fmt.Errorf("add flag %q: %w", flag, err) - } - } - case port.FlagOpDel: - for _, flag := range flags { - if _, err := tx.Exec(ctx, ` - DELETE FROM message_flags WHERE (message_id, received_at) IN ( - SELECT id, received_at FROM email_messages WHERE folder_id = $1 AND uid = $2 AND expunged_at IS NULL - ) AND flag = $3 - `, folderID, uid, flag); err != nil { - return false, fmt.Errorf("delete flag %q: %w", flag, err) - } - } - case port.FlagOpSet: - if _, err := tx.Exec(ctx, ` - DELETE FROM message_flags WHERE (message_id, received_at) IN ( - SELECT id, received_at FROM email_messages WHERE folder_id = $1 AND uid = $2 AND expunged_at IS NULL - ) - `, folderID, uid); err != nil { - return false, fmt.Errorf("clear flags before set: %w", err) - } - for _, flag := range flags { - if _, err := tx.Exec(ctx, ` - INSERT INTO message_flags (message_id, received_at, flag) - SELECT id, received_at, $3 FROM email_messages WHERE folder_id = $1 AND uid = $2 AND expunged_at IS NULL - ON CONFLICT DO NOTHING - `, folderID, uid, flag); err != nil { - return false, fmt.Errorf("set flag %q: %w", flag, err) - } - } - } - - if err := tx.Commit(ctx); err != nil { - return false, fmt.Errorf("commit: %w", err) - } - return true, nil + return r.store.SetFlags(ctx, folderID, uid, op, flags, unchangedSince) } diff --git a/services/protocols/internal/infrastructure/postgres/mail_store.go b/services/protocols/internal/infrastructure/postgres/mail_store.go new file mode 100644 index 0000000..f529dc5 --- /dev/null +++ b/services/protocols/internal/infrastructure/postgres/mail_store.go @@ -0,0 +1,451 @@ +package postgres + +import ( + "context" + "errors" + "fmt" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" + + "lambdamail/protocols/internal/application/port" +) + +// MailStore is the one place that changes the state of messages and folders. +// +// IMAP, POP3, the webmail and the server's own filing all end up changing the +// same rows, and those rows carry cached numbers that have to stay true: the +// folder's unread and total counters, its highest modseq (what a CONDSTORE +// client uses to learn that something changed), the next UID, and the +// mailbox's used bytes. When each path implemented this on its own they +// drifted apart - reading over IMAP never moved the unread counter, deleting in +// the webmail never gave the quota back, a repeated expunge could push a +// counter below zero - and every fix had to be made twice, which is how one of +// the two went missing. +// +// So each state change is written here, once, in one transaction, and every +// path calls it. The repositories IMAP and the webmail used are now thin +// adapters over this: they translate their way of naming things (a folder id +// and a UID, or a mailbox, a folder name and a UID) and nothing else. +// +// Rules held for every method: +// +// - A counter never goes below zero, so a double operation cannot corrupt it. +// - An operation that changes what a client can observe raises the folder's +// modseq; one that changes nothing does not. +// - A message is located by folder, so another mailbox's mail cannot be +// reached by guessing a UID. +type MailStore struct { + pool *pgxpool.Pool +} + +func NewMailStore(pool *pgxpool.Pool) *MailStore { + return &MailStore{pool: pool} +} + +// messageRow is the slice of a message every operation needs. +type messageRow struct { + ID uuid.UUID + ReceivedAt any + MailboxID string + Size int64 + ModSeq uint64 + Seen bool +} + +// loadMessage finds a live message by folder and UID and locks it, so two +// concurrent changes to the same message cannot both read the old state. +func loadMessage(ctx context.Context, tx pgx.Tx, folderID string, uid uint32) (*messageRow, error) { + var m messageRow + err := tx.QueryRow(ctx, ` + SELECT e.id, e.received_at, e.mailbox_id::text, e.size_bytes, e.modseq, + EXISTS (SELECT 1 FROM message_flags mf + WHERE mf.message_id = e.id AND mf.received_at = e.received_at AND mf.flag = '\Seen') + FROM email_messages e + WHERE e.folder_id = $1 AND e.uid = $2 AND e.expunged_at IS NULL + FOR UPDATE + `, folderID, int64(uid)).Scan(&m.ID, &m.ReceivedAt, &m.MailboxID, &m.Size, &m.ModSeq, &m.Seen) + if errors.Is(err, pgx.ErrNoRows) { + return nil, ErrMessageNotFound + } + if err != nil { + return nil, fmt.Errorf("find message folder=%s uid=%d: %w", folderID, uid, err) + } + return &m, nil +} + +// ResolveFolder finds a folder of a mailbox by its special-use role or by its +// name, the way the webmail and IMAP clients both address one ("trash" and +// "Trash" reach the same place; a folder the user made has no role and is +// reached by name). port.ErrFolderMissing when there is none. +func (s *MailStore) ResolveFolder(ctx context.Context, mailboxID, ref string) (string, error) { + var id string + err := s.pool.QueryRow(ctx, ` + SELECT id::text FROM folders + WHERE mailbox_id = $1 AND (special_use = LOWER($2) OR LOWER(name) = LOWER($2)) + ORDER BY special_use IS NOT NULL DESC + LIMIT 1 + `, mailboxID, ref).Scan(&id) + if errors.Is(err, pgx.ErrNoRows) { + return "", port.ErrFolderMissing + } + return id, err +} + +// SetFlags applies op to one message's flags. Add and Del touch individual +// flags, Set replaces the whole set. If unchangedSince > 0 and the message has +// changed since, nothing is written and updated is false (RFC 7162 +// UNCHANGEDSINCE). +// +// \Seen is the flag with a side effect: the folder's unread counter follows it, +// in the same transaction, by looking at the flag before and after rather than +// at which operation was asked for. That is what keeps STORE +FLAGS, STORE +// -FLAGS, STORE FLAGS (replace) and the webmail's mark-as-read all counting the +// same way. +func (s *MailStore) SetFlags(ctx context.Context, folderID string, uid uint32, op port.FlagOp, flags []string, unchangedSince uint64) (bool, error) { + tx, err := s.pool.Begin(ctx) + if err != nil { + return false, fmt.Errorf("begin transaction: %w", err) + } + defer tx.Rollback(ctx) + + msg, err := loadMessage(ctx, tx, folderID, uid) + if err != nil { + return false, err + } + + if unchangedSince > 0 && msg.ModSeq > unchangedSince { + return false, nil + } + + before, err := flagSet(ctx, tx, msg) + if err != nil { + return false, err + } + + switch op { + case port.FlagOpAdd: + for _, flag := range flags { + if _, err := tx.Exec(ctx, ` + INSERT INTO message_flags (message_id, received_at, flag) VALUES ($1, $2, $3) + ON CONFLICT DO NOTHING`, msg.ID, msg.ReceivedAt, flag); err != nil { + return false, fmt.Errorf("add flag %q: %w", flag, err) + } + } + case port.FlagOpDel: + for _, flag := range flags { + if _, err := tx.Exec(ctx, ` + DELETE FROM message_flags WHERE message_id = $1 AND received_at = $2 AND flag = $3`, + msg.ID, msg.ReceivedAt, flag); err != nil { + return false, fmt.Errorf("delete flag %q: %w", flag, err) + } + } + case port.FlagOpSet: + if _, err := tx.Exec(ctx, ` + DELETE FROM message_flags WHERE message_id = $1 AND received_at = $2`, msg.ID, msg.ReceivedAt); err != nil { + return false, fmt.Errorf("clear flags before set: %w", err) + } + for _, flag := range flags { + if _, err := tx.Exec(ctx, ` + INSERT INTO message_flags (message_id, received_at, flag) VALUES ($1, $2, $3) + ON CONFLICT DO NOTHING`, msg.ID, msg.ReceivedAt, flag); err != nil { + return false, fmt.Errorf("set flag %q: %w", flag, err) + } + } + } + + after, err := flagSet(ctx, tx, msg) + if err != nil { + return false, err + } + + // Nothing changed: no modseq, no counter. A client that stores a flag the + // message already has must not be told the folder changed. + if sameFlags(before, after) { + return true, tx.Commit(ctx) + } + + unreadDelta := 0 + switch { + case before[`\Seen`] && !after[`\Seen`]: + unreadDelta = 1 + case !before[`\Seen`] && after[`\Seen`]: + unreadDelta = -1 + } + + var modseq int64 + if err := tx.QueryRow(ctx, ` + UPDATE folders + SET highest_modseq = highest_modseq + 1, + unread_count = GREATEST(0, unread_count + $2) + WHERE id = $1 + RETURNING highest_modseq`, folderID, unreadDelta).Scan(&modseq); err != nil { + return false, fmt.Errorf("advance folder modseq: %w", err) + } + if _, err := tx.Exec(ctx, `UPDATE email_messages SET modseq = $3 WHERE id = $1 AND received_at = $2`, + msg.ID, msg.ReceivedAt, modseq); err != nil { + return false, fmt.Errorf("update message modseq: %w", err) + } + + return true, tx.Commit(ctx) +} + +func flagSet(ctx context.Context, tx pgx.Tx, msg *messageRow) (map[string]bool, error) { + rows, err := tx.Query(ctx, `SELECT flag FROM message_flags WHERE message_id = $1 AND received_at = $2`, msg.ID, msg.ReceivedAt) + if err != nil { + return nil, fmt.Errorf("read flags: %w", err) + } + defer rows.Close() + set := map[string]bool{} + for rows.Next() { + var f string + if err := rows.Scan(&f); err != nil { + return nil, err + } + set[f] = true + } + return set, rows.Err() +} + +func sameFlags(a, b map[string]bool) bool { + if len(a) != len(b) { + return false + } + for k := range a { + if !b[k] { + return false + } + } + return true +} + +// Expunge removes messages from a folder (soft delete: the row stays, marked, +// and the blob stays referenced until the retention sweep, so a mistake is +// recoverable and a client holding the UID still resolves it). +// +// For each message that was really there it lowers the folder's total, lowers +// unread if the message was unread, and gives its bytes back to the mailbox's +// quota. A message that is already gone is skipped, touching nothing, which is +// what makes a repeated expunge safe. It returns how many messages were +// removed. +func (s *MailStore) Expunge(ctx context.Context, folderID string, uids []uint32) (int, error) { + if len(uids) == 0 { + return 0, nil + } + tx, err := s.pool.Begin(ctx) + if err != nil { + return 0, fmt.Errorf("begin transaction: %w", err) + } + defer tx.Rollback(ctx) + + removed := 0 + for _, uid := range uids { + msg, err := loadMessage(ctx, tx, folderID, uid) + if errors.Is(err, ErrMessageNotFound) { + continue + } + if err != nil { + return 0, err + } + + if _, err := tx.Exec(ctx, `UPDATE email_messages SET expunged_at = NOW() WHERE id = $1 AND received_at = $2`, + msg.ID, msg.ReceivedAt); err != nil { + return 0, fmt.Errorf("expunge uid %d: %w", uid, err) + } + + unreadDelta := 0 + if !msg.Seen { + unreadDelta = -1 + } + if _, err := tx.Exec(ctx, ` + UPDATE folders + SET total_count = GREATEST(0, total_count - 1), + unread_count = GREATEST(0, unread_count + $2), + highest_modseq = highest_modseq + 1 + WHERE id = $1`, folderID, unreadDelta); err != nil { + return 0, fmt.Errorf("lower folder counters for uid %d: %w", uid, err) + } + + if _, err := tx.Exec(ctx, `UPDATE mailboxes SET used_bytes = GREATEST(0, used_bytes - $1) WHERE id = $2`, + msg.Size, msg.MailboxID); err != nil { + return 0, fmt.Errorf("give quota back for uid %d: %w", uid, err) + } + removed++ + } + + return removed, tx.Commit(ctx) +} + +// Move relocates one message to another folder of the same mailbox and returns +// the UID it has there. The message keeps its identity, its flags and its blob; +// only the folder and the UID change, because a UID means something only inside +// one folder. The quota does not move: the bytes stay in the same mailbox. +func (s *MailStore) Move(ctx context.Context, sourceFolderID string, uid uint32, destFolderID string) (uint32, error) { + tx, err := s.pool.Begin(ctx) + if err != nil { + return 0, err + } + defer tx.Rollback(ctx) + + msg, err := loadMessage(ctx, tx, sourceFolderID, uid) + if err != nil { + return 0, err + } + + var destUID int64 + if err := tx.QueryRow(ctx, `SELECT uid_next FROM folders WHERE id = $1 FOR UPDATE`, destFolderID).Scan(&destUID); err != nil { + return 0, fmt.Errorf("lock destination folder: %w", err) + } + + unread := 0 + if !msg.Seen { + unread = 1 + } + + var modseq int64 + if err := tx.QueryRow(ctx, ` + UPDATE folders + SET uid_next = uid_next + 1, highest_modseq = highest_modseq + 1, + total_count = total_count + 1, unread_count = unread_count + $2 + WHERE id = $1 + RETURNING highest_modseq`, destFolderID, unread).Scan(&modseq); err != nil { + return 0, fmt.Errorf("advance destination folder: %w", err) + } + + if _, err := tx.Exec(ctx, `UPDATE email_messages SET folder_id = $3, uid = $4, modseq = $5 WHERE id = $1 AND received_at = $2`, + msg.ID, msg.ReceivedAt, destFolderID, destUID, modseq); err != nil { + return 0, fmt.Errorf("move message: %w", err) + } + + if _, err := tx.Exec(ctx, ` + UPDATE folders + SET total_count = GREATEST(0, total_count - 1), + unread_count = GREATEST(0, unread_count - $2), + highest_modseq = highest_modseq + 1 + WHERE id = $1`, sourceFolderID, unread); err != nil { + return 0, fmt.Errorf("lower source folder counters: %w", err) + } + + return uint32(destUID), tx.Commit(ctx) +} + +// Copy duplicates messages into another folder. Each copy is its own message +// (own UID, own flags copied from the original) that shares the original's +// blob, and it counts against the quota like any other message. One transaction +// per message, so a failure part-way leaves the earlier copies whole. +func (s *MailStore) Copy(ctx context.Context, sourceFolderID string, uids []uint32, destFolderID string) ([]port.CopiedMessage, error) { + results := make([]port.CopiedMessage, 0, len(uids)) + for _, uid := range uids { + destUID, err := s.copyOne(ctx, sourceFolderID, uid, destFolderID) + if err != nil { + return nil, fmt.Errorf("copy uid %d: %w", uid, err) + } + results = append(results, port.CopiedMessage{SourceUID: uid, DestUID: destUID}) + } + return results, nil +} + +func (s *MailStore) copyOne(ctx context.Context, sourceFolderID string, sourceUID uint32, destFolderID string) (uint32, error) { + tx, err := s.pool.Begin(ctx) + if err != nil { + return 0, fmt.Errorf("begin transaction: %w", err) + } + defer tx.Rollback(ctx) + + src, err := loadMessage(ctx, tx, sourceFolderID, sourceUID) + if err != nil { + return 0, err + } + + var destUID int64 + var destMailboxID string + if err := tx.QueryRow(ctx, `SELECT uid_next, mailbox_id::text FROM folders WHERE id = $1 FOR UPDATE`, destFolderID).Scan(&destUID, &destMailboxID); err != nil { + return 0, fmt.Errorf("lock destination folder: %w", err) + } + + unread := 0 + if !src.Seen { + unread = 1 + } + var modseq int64 + if err := tx.QueryRow(ctx, ` + UPDATE folders + SET uid_next = uid_next + 1, highest_modseq = highest_modseq + 1, + total_count = total_count + 1, unread_count = unread_count + $2 + WHERE id = $1 + RETURNING highest_modseq`, destFolderID, unread).Scan(&modseq); err != nil { + return 0, fmt.Errorf("advance destination folder: %w", err) + } + + var newID uuid.UUID + var newReceivedAt any + if err := tx.QueryRow(ctx, ` + INSERT INTO email_messages ( + mailbox_id, folder_id, uid, modseq, blob_id, + sender_address, recipient_addresses, size_bytes, + spf_result, dkim_result, dmarc_result, + from_display_name, subject, snippet, message_id_header, in_reply_to, thread_id, has_attachments, + arc_result, dane_status, tls_version, spam_score, spam_verdict, virus_name + ) + SELECT $2::uuid, $3::uuid, $4::bigint, $5::bigint, blob_id, + sender_address, recipient_addresses, size_bytes, + spf_result, dkim_result, dmarc_result, + from_display_name, subject, snippet, message_id_header, in_reply_to, thread_id, has_attachments, + arc_result, dane_status, tls_version, spam_score, spam_verdict, virus_name + FROM email_messages WHERE id = $1 AND received_at = $6 + RETURNING id, received_at`, + src.ID, destMailboxID, destFolderID, destUID, modseq, src.ReceivedAt).Scan(&newID, &newReceivedAt); err != nil { + return 0, fmt.Errorf("insert copied message: %w", err) + } + + if _, err := tx.Exec(ctx, `UPDATE message_blobs SET ref_count = ref_count + 1 WHERE id = (SELECT blob_id FROM email_messages WHERE id = $1 AND received_at = $2)`, + newID, newReceivedAt); err != nil { + return 0, fmt.Errorf("share the blob: %w", err) + } + if _, err := tx.Exec(ctx, `UPDATE mailboxes SET used_bytes = used_bytes + $1 WHERE id = $2`, src.Size, destMailboxID); err != nil { + return 0, fmt.Errorf("count the copy against the quota: %w", err) + } + if _, err := tx.Exec(ctx, ` + INSERT INTO message_flags (message_id, received_at, flag) + SELECT $1, $2, flag FROM message_flags WHERE message_id = $3 AND received_at = $4`, + newID, newReceivedAt, src.ID, src.ReceivedAt); err != nil { + return 0, fmt.Errorf("copy flags: %w", err) + } + + return uint32(destUID), tx.Commit(ctx) +} + +// DeleteFolder removes a folder the user made, with the mail in it. The +// messages are expunged rather than deleted (recoverable, like any expunge) and +// their bytes go back to the quota. +func (s *MailStore) DeleteFolder(ctx context.Context, folderID string) error { + tx, err := s.pool.Begin(ctx) + if err != nil { + return err + } + defer tx.Rollback(ctx) + + var mailboxID string + if err := tx.QueryRow(ctx, `SELECT mailbox_id::text FROM folders WHERE id = $1 FOR UPDATE`, folderID).Scan(&mailboxID); err != nil { + return fmt.Errorf("lock folder: %w", err) + } + + var freed int64 + if err := tx.QueryRow(ctx, ` + WITH gone AS ( + UPDATE email_messages SET expunged_at = NOW() + WHERE folder_id = $1 AND expunged_at IS NULL + RETURNING size_bytes) + SELECT COALESCE(SUM(size_bytes), 0) FROM gone`, folderID).Scan(&freed); err != nil { + return fmt.Errorf("expunge the folder's messages: %w", err) + } + + if _, err := tx.Exec(ctx, `UPDATE mailboxes SET used_bytes = GREATEST(0, used_bytes - $1) WHERE id = $2`, freed, mailboxID); err != nil { + return fmt.Errorf("give quota back: %w", err) + } + if _, err := tx.Exec(ctx, `DELETE FROM folders WHERE id = $1`, folderID); err != nil { + return fmt.Errorf("delete folder: %w", err) + } + return tx.Commit(ctx) +} diff --git a/services/protocols/internal/infrastructure/postgres/webmail_repository.go b/services/protocols/internal/infrastructure/postgres/webmail_repository.go index f39c8ca..e4adef9 100644 --- a/services/protocols/internal/infrastructure/postgres/webmail_repository.go +++ b/services/protocols/internal/infrastructure/postgres/webmail_repository.go @@ -19,11 +19,12 @@ var ErrMessageNotFound = errors.New("webmail: message not found") // WebmailRepository implements the webmail read side against Postgres. type WebmailRepository struct { - pool *pgxpool.Pool + pool *pgxpool.Pool + store *MailStore } func NewWebmailRepository(pool *pgxpool.Pool) *WebmailRepository { - return &WebmailRepository{pool: pool} + return &WebmailRepository{pool: pool, store: NewMailStore(pool)} } func (r *WebmailRepository) FindMailboxIDByAddress(ctx context.Context, address string) (string, error) { @@ -157,318 +158,96 @@ func (r *WebmailRepository) GetMessageBlob( return blobID, err } -// MarkSeen keeps the \Seen flag and the folder's unread counter in step. They -// are updated together in one transaction, because a counter that disagrees -// with the flags is what makes an inbox show unread mail that is not there. -func (r *WebmailRepository) MarkSeen( - ctx context.Context, mailboxID, folderName string, uid uint32, seen bool, -) error { - tx, err := r.pool.Begin(ctx) +// The methods below change state. They do not do it themselves: the webmail +// names a message by mailbox, folder name and UID, MailStore by folder id and +// UID, and these translate. Every counter, modseq and quota rule lives in +// MailStore, which IMAP uses too, so the two can no longer disagree. + +// messageFolder resolves the folder a message is addressed in. A folder that is +// not there and a message that is not there look the same to the caller, so a +// stranger cannot probe which folders exist. +func (r *WebmailRepository) messageFolder(ctx context.Context, mailboxID, folderName string) (string, error) { + id, err := r.store.ResolveFolder(ctx, mailboxID, folderName) + if errors.Is(err, port.ErrFolderMissing) { + return "", ErrMessageNotFound + } + return id, err +} + +// MarkSeen sets or clears \Seen. The unread counter follows inside MailStore. +func (r *WebmailRepository) MarkSeen(ctx context.Context, mailboxID, folderName string, uid uint32, seen bool) error { + return r.setFlag(ctx, mailboxID, folderName, uid, `\Seen`, seen) +} + +// MarkFlagged sets or clears \Flagged. +func (r *WebmailRepository) MarkFlagged(ctx context.Context, mailboxID, folderName string, uid uint32, flagged bool) error { + return r.setFlag(ctx, mailboxID, folderName, uid, `\Flagged`, flagged) +} + +func (r *WebmailRepository) setFlag(ctx context.Context, mailboxID, folderName string, uid uint32, flag string, on bool) error { + folderID, err := r.messageFolder(ctx, mailboxID, folderName) if err != nil { return err } - defer tx.Rollback(ctx) + op := port.FlagOpAdd + if !on { + op = port.FlagOpDel + } + _, err = r.store.SetFlags(ctx, folderID, uid, op, []string{flag}, 0) + return err +} - var messageID uuid.UUID - var receivedAt any - var folderID string - err = tx.QueryRow(ctx, ` - SELECT e.id, e.received_at, f.id::text FROM email_messages e - JOIN folders f ON f.id = e.folder_id - WHERE e.mailbox_id = $1 AND `+folderCondition+` - AND e.uid = $3 AND e.expunged_at IS NULL - `, mailboxID, folderName, int64(uid)).Scan(&messageID, &receivedAt, &folderID) - if errors.Is(err, pgx.ErrNoRows) { +// Expunge removes one message for good (soft delete) and gives its bytes back. +func (r *WebmailRepository) Expunge(ctx context.Context, mailboxID, folderName string, uid uint32) error { + folderID, err := r.messageFolder(ctx, mailboxID, folderName) + if err != nil { + return err + } + removed, err := r.store.Expunge(ctx, folderID, []uint32{uid}) + if err != nil { + return err + } + if removed == 0 { return ErrMessageNotFound } - if err != nil { - return err - } - - var changed int64 - if seen { - tag, err := tx.Exec(ctx, ` - INSERT INTO message_flags (message_id, received_at, flag) - VALUES ($1, $2, '\Seen') ON CONFLICT DO NOTHING - `, messageID, receivedAt) - if err != nil { - return err - } - changed = tag.RowsAffected() - } else { - tag, err := tx.Exec(ctx, ` - DELETE FROM message_flags WHERE message_id = $1 AND received_at = $2 AND flag = '\Seen' - `, messageID, receivedAt) - if err != nil { - return err - } - changed = tag.RowsAffected() - } - - // Only a real transition moves the counter, so opening an already-read - // message repeatedly cannot drive unread_count negative. - if changed > 0 { - delta := 1 - if seen { - delta = -1 - } - if _, err := tx.Exec(ctx, ` - UPDATE folders SET unread_count = GREATEST(0, unread_count + $2), highest_modseq = highest_modseq + 1 - WHERE id = $1 - `, folderID, delta); err != nil { - return err - } - } - - return tx.Commit(ctx) -} - -// MarkFlagged sets or clears \Flagged. Unlike \Seen it moves no counter, but a -// real change still bumps the folder's modseq: that is how an IMAP client -// holding the folder open learns the flag changed without a full resync. -func (r *WebmailRepository) MarkFlagged( - ctx context.Context, mailboxID, folderName string, uid uint32, flagged bool, -) error { - tx, err := r.pool.Begin(ctx) - if err != nil { - return err - } - defer tx.Rollback(ctx) - - var messageID uuid.UUID - var receivedAt any - var folderID string - err = tx.QueryRow(ctx, ` - SELECT e.id, e.received_at, f.id::text FROM email_messages e - JOIN folders f ON f.id = e.folder_id - WHERE e.mailbox_id = $1 AND `+folderCondition+` - AND e.uid = $3 AND e.expunged_at IS NULL - `, mailboxID, folderName, int64(uid)).Scan(&messageID, &receivedAt, &folderID) - if errors.Is(err, pgx.ErrNoRows) { - return ErrMessageNotFound - } - if err != nil { - return err - } - - var changed int64 - if flagged { - tag, err := tx.Exec(ctx, ` - INSERT INTO message_flags (message_id, received_at, flag) - VALUES ($1, $2, '\Flagged') ON CONFLICT DO NOTHING - `, messageID, receivedAt) - if err != nil { - return err - } - changed = tag.RowsAffected() - } else { - tag, err := tx.Exec(ctx, ` - DELETE FROM message_flags WHERE message_id = $1 AND received_at = $2 AND flag = '\Flagged' - `, messageID, receivedAt) - if err != nil { - return err - } - changed = tag.RowsAffected() - } - - if changed > 0 { - if _, err := tx.Exec(ctx, - `UPDATE folders SET highest_modseq = highest_modseq + 1 WHERE id = $1`, folderID); err != nil { - return err - } - } - return tx.Commit(ctx) -} - -// Expunge marks one message deleted and keeps the folder counters in step. -// -// Soft delete, matching the expunged_at column the rest of the schema uses: the -// blob stays referenced until the retention sweep, so a mistaken delete is -// recoverable and IMAP clients holding the UID still resolve it. -// -// The message is located by mailbox as well as UID, so a caller cannot expunge -// somebody else's mail by guessing a number. -func (r *WebmailRepository) Expunge( - ctx context.Context, mailboxID, folderName string, uid uint32, -) error { - tx, err := r.pool.Begin(ctx) - if err != nil { - return err - } - defer tx.Rollback(ctx) - - var messageID uuid.UUID - var folderID string - var wasUnread bool - err = tx.QueryRow(ctx, ` - SELECT e.id, f.id::text, - NOT EXISTS (SELECT 1 FROM message_flags mf - WHERE mf.message_id = e.id - AND mf.received_at = e.received_at - AND mf.flag = '\Seen') - FROM email_messages e - JOIN folders f ON f.id = e.folder_id - WHERE e.mailbox_id = $1 AND `+folderCondition+` - AND e.uid = $3 AND e.expunged_at IS NULL - `, mailboxID, folderName, int64(uid)).Scan(&messageID, &folderID, &wasUnread) - if errors.Is(err, pgx.ErrNoRows) { - return ErrMessageNotFound - } - if err != nil { - return err - } - - if _, err := tx.Exec(ctx, - `UPDATE email_messages SET expunged_at = NOW() WHERE id = $1`, messageID); err != nil { - return err - } - - // GREATEST guards the counters against ever going negative, which a - // double expunge of the same row would otherwise cause. - unreadDelta := 0 - if wasUnread { - unreadDelta = -1 - } - if _, err := tx.Exec(ctx, ` - UPDATE folders - SET total_count = GREATEST(0, total_count - 1), - unread_count = GREATEST(0, unread_count + $2), - highest_modseq = highest_modseq + 1 - WHERE id = $1 - `, folderID, unreadDelta); err != nil { - return err - } - - return tx.Commit(ctx) -} - -// MoveToTrash relocates one message into the mailbox's Trash folder and -// returns the UID it was given there. -// -// This is what "delete" means in a mail client, and the webmail had no way to -// do it at all: the only removal in the codebase was Expunge, reachable solely -// when an autosave superseded a draft. A message the user wanted rid of - the -// empty draft left behind by a sent message, most visibly - could not be -// removed from any screen. -// -// The message keeps its identity and its blob; only the folder and the UID -// change, because a UID is meaningful only within one folder. Deleting from -// Trash itself is Expunge's job, not this one's. -func (r *WebmailRepository) MoveToTrash( - ctx context.Context, mailboxID, folderName string, uid uint32, -) (uint32, error) { - return r.moveInto(ctx, mailboxID, folderName, uid, trashTarget) -} - -// MoveToFolder files one message into a named folder. -// -// The target is matched on its special-use role or its name, the same way the -// rest of this repository addresses folders, so "archive" and "Archive" both -// reach the same place and a folder the user created themselves - which has no -// role - is reachable by name. -func (r *WebmailRepository) MoveToFolder( - ctx context.Context, mailboxID, folderName string, uid uint32, target string, -) (uint32, error) { - return r.moveInto(ctx, mailboxID, folderName, uid, target) + return nil } // trashTarget is what MoveToTrash resolves; kept as a constant so the two // callers cannot drift. const trashTarget = "trash" -func (r *WebmailRepository) moveInto( - ctx context.Context, mailboxID, folderName string, uid uint32, target string, -) (uint32, error) { - tx, err := r.pool.Begin(ctx) - if err != nil { - return 0, err - } - defer tx.Rollback(ctx) - - var messageID uuid.UUID - var sourceFolderID string - var wasUnread bool - err = tx.QueryRow(ctx, ` - SELECT e.id, f.id::text, - NOT EXISTS (SELECT 1 FROM message_flags mf - WHERE mf.message_id = e.id - AND mf.received_at = e.received_at - AND mf.flag = '\Seen') - FROM email_messages e - JOIN folders f ON f.id = e.folder_id - WHERE e.mailbox_id = $1 AND `+folderCondition+` - AND e.uid = $3 AND e.expunged_at IS NULL - `, mailboxID, folderName, int64(uid)).Scan(&messageID, &sourceFolderID, &wasUnread) - if errors.Is(err, pgx.ErrNoRows) { - return 0, ErrMessageNotFound - } - if err != nil { - return 0, err - } - - // Deleting something already in Trash has no further destination, so the - // caller is told to expunge instead of the message being moved onto itself. - var trashFolderID string - var trashUID int64 - err = tx.QueryRow(ctx, ` - SELECT id::text, uid_next FROM folders - WHERE mailbox_id = $1 AND (special_use = LOWER($2) OR LOWER(name) = LOWER($2)) - ORDER BY special_use IS NOT NULL DESC - LIMIT 1 - FOR UPDATE - `, mailboxID, target).Scan(&trashFolderID, &trashUID) - if errors.Is(err, pgx.ErrNoRows) { - return 0, port.ErrNoTrashFolder - } - if err != nil { - return 0, err - } - if trashFolderID == sourceFolderID { - return 0, port.ErrAlreadyInTrash - } - - var modseq int64 - if err := tx.QueryRow(ctx, ` - UPDATE folders SET uid_next = uid_next + 1, highest_modseq = highest_modseq + 1, - total_count = total_count + 1, unread_count = unread_count + $2 - WHERE id = $1 - RETURNING highest_modseq - `, trashFolderID, boolToInt(wasUnread)).Scan(&modseq); err != nil { - return 0, err - } - - if _, err := tx.Exec(ctx, ` - UPDATE email_messages SET folder_id = $2, uid = $3, modseq = $4 - WHERE id = $1 - `, messageID, trashFolderID, trashUID, modseq); err != nil { - return 0, err - } - - unreadDelta := 0 - if wasUnread { - unreadDelta = -1 - } - if _, err := tx.Exec(ctx, ` - UPDATE folders - SET total_count = GREATEST(0, total_count - 1), - unread_count = GREATEST(0, unread_count + $2), - highest_modseq = highest_modseq + 1 - WHERE id = $1 - `, sourceFolderID, unreadDelta); err != nil { - return 0, err - } - - if err := tx.Commit(ctx); err != nil { - return 0, err - } - return uint32(trashUID), nil +// MoveToTrash is what "delete" means in a mail client: the message goes to the +// Trash folder and keeps its identity. Deleting from Trash itself is Expunge's +// job, so asking to move out of Trash reports port.ErrAlreadyInTrash. +func (r *WebmailRepository) MoveToTrash(ctx context.Context, mailboxID, folderName string, uid uint32) (uint32, error) { + return r.moveInto(ctx, mailboxID, folderName, uid, trashTarget) } -func boolToInt(b bool) int { - if b { - return 1 +// MoveToFolder files one message into a named folder, matched by role or name. +func (r *WebmailRepository) MoveToFolder(ctx context.Context, mailboxID, folderName string, uid uint32, target string) (uint32, error) { + return r.moveInto(ctx, mailboxID, folderName, uid, target) +} + +func (r *WebmailRepository) moveInto(ctx context.Context, mailboxID, folderName string, uid uint32, target string) (uint32, error) { + sourceID, err := r.messageFolder(ctx, mailboxID, folderName) + if err != nil { + return 0, err } - return 0 + destID, err := r.store.ResolveFolder(ctx, mailboxID, target) + if errors.Is(err, port.ErrFolderMissing) { + if target == trashTarget { + return 0, port.ErrNoTrashFolder + } + return 0, port.ErrFolderMissing + } + if err != nil { + return 0, err + } + if destID == sourceID { + return 0, port.ErrAlreadyInTrash + } + return r.store.Move(ctx, sourceID, uid, destID) } // --- folders the mailbox owner creates for themselves -------------------- @@ -507,20 +286,12 @@ func (r *WebmailRepository) RenameFolder(ctx context.Context, mailboxID, from, t return nil } -// DeleteFolder removes a folder and the mail filed in it. -// -// The messages are expunged rather than deleted outright, matching what the -// rest of the schema does: the blobs stay referenced until the retention sweep, -// so a folder deleted by mistake is still recoverable from the database. +// DeleteFolder removes a folder the owner made, and the mail filed in it. The +// messages are expunged (recoverable) and their bytes go back to the quota, +// both inside MailStore. func (r *WebmailRepository) DeleteFolder(ctx context.Context, mailboxID, name string) error { - tx, err := r.pool.Begin(ctx) - if err != nil { - return err - } - defer tx.Rollback(ctx) - var folderID string - err = tx.QueryRow(ctx, ` + err := r.pool.QueryRow(ctx, ` SELECT id::text FROM folders WHERE mailbox_id = $1 AND LOWER(name) = LOWER($2) AND special_use IS NULL `, mailboxID, name).Scan(&folderID) @@ -530,16 +301,5 @@ func (r *WebmailRepository) DeleteFolder(ctx context.Context, mailboxID, name st if err != nil { return err } - - if _, err := tx.Exec(ctx, ` - UPDATE email_messages SET expunged_at = NOW() - WHERE folder_id = $1 AND expunged_at IS NULL - `, folderID); err != nil { - return fmt.Errorf("expunge the folder's messages: %w", err) - } - - if _, err := tx.Exec(ctx, `DELETE FROM folders WHERE id = $1`, folderID); err != nil { - return fmt.Errorf("delete folder %q: %w", name, err) - } - return tx.Commit(ctx) + return r.store.DeleteFolder(ctx, folderID) } From a8941a61912d0a36198ea82872530fa041dfb6fa Mon Sep 17 00:00:00 2001 From: Lucas Rayan Date: Fri, 16 Oct 2026 13:19:55 -0300 Subject: [PATCH 3/6] refactor(mail): one definition of the standard folders, and reconcile the drift The console created Reports, the CLI that makes the first administrator did not, so that account had no folder for DMARC and TLS-RPT reports. The list now lives in one SQL function, provision_default_folders (migration 0014), which both call. The same migration reconciles what the two paths had left wrong in existing mailboxes: folder counters and the mailbox's used bytes are recomputed from the rows (only where they differ, so it is safe to run again). create-admin can now be tested: fail() throws instead of exiting, and the command runs only when invoked as one. The new test fails on the old code (no Reports). --- .../0014_mailbox_state_single_source.down.sql | 4 + .../0014_mailbox_state_single_source.up.sql | 77 +++++++++ services/auth/src/cli.test.ts | 46 +++++ services/auth/src/cli.ts | 47 +++--- services/auth/src/repository.ts | 15 +- .../postgres/mailbox_state_migration_test.go | 158 ++++++++++++++++++ 6 files changed, 310 insertions(+), 37 deletions(-) create mode 100644 migrations/0014_mailbox_state_single_source.down.sql create mode 100644 migrations/0014_mailbox_state_single_source.up.sql create mode 100644 services/auth/src/cli.test.ts create mode 100644 services/protocols/internal/infrastructure/postgres/mailbox_state_migration_test.go diff --git a/migrations/0014_mailbox_state_single_source.down.sql b/migrations/0014_mailbox_state_single_source.down.sql new file mode 100644 index 0000000..9dfb621 --- /dev/null +++ b/migrations/0014_mailbox_state_single_source.down.sql @@ -0,0 +1,4 @@ +-- The reconciliation corrected cached numbers and cannot be undone (and has no +-- reason to be); only the function is removed. Folders it added stay: they are +-- ordinary folders by now and may hold mail. +DROP FUNCTION IF EXISTS provision_default_folders(UUID); diff --git a/migrations/0014_mailbox_state_single_source.up.sql b/migrations/0014_mailbox_state_single_source.up.sql new file mode 100644 index 0000000..98d52d4 --- /dev/null +++ b/migrations/0014_mailbox_state_single_source.up.sql @@ -0,0 +1,77 @@ +-- One definition of "a mailbox's standard folders", and a reconciliation of the +-- cached numbers that two code paths used to keep on their own. +-- +-- 1. provision_default_folders +-- +-- The standard folders were created in two places that had already diverged: +-- the console's createMailbox added Reports, the CLI that creates the first +-- administrator did not, so that account had no folder for DMARC and TLS-RPT +-- reports and they fell into its INBOX. Both now call this function, so the +-- list lives here and nowhere else. It is idempotent: it adds only what is +-- missing, matching by name without regard to case. +CREATE OR REPLACE FUNCTION provision_default_folders(p_mailbox_id UUID) +RETURNS VOID +LANGUAGE sql +AS $$ + INSERT INTO folders (mailbox_id, name, special_use) + SELECT p_mailbox_id, f.name, f.special_use + FROM (VALUES ('INBOX', 'inbox'), + ('Sent', 'sent'), + ('Drafts', 'drafts'), + ('Trash', 'trash'), + ('Junk', 'junk'), + ('Archive', 'archive'), + -- Where delivered DMARC and TLS-RPT reports are filed. IMAP has no + -- special-use role for these, so it is matched by name. + ('Reports', NULL)) AS f(name, special_use) + WHERE NOT EXISTS ( + SELECT 1 FROM folders x + WHERE x.mailbox_id = p_mailbox_id + AND x.parent_id IS NULL + AND LOWER(x.name) = LOWER(f.name)); +$$; + +-- Mailboxes created before the function that lack a standard folder get it now. +SELECT provision_default_folders(id) FROM mailboxes; + +-- 2. Reconcile the cached numbers with the rows. +-- +-- Folder counters, and the mailbox's used bytes, are denormalised caches. IMAP +-- and the webmail each maintained them and each got something wrong (IMAP never +-- moved the unread counter; the webmail never gave the quota back), so existing +-- mailboxes carry the drift. Every path now goes through one MailStore that +-- keeps them true; this corrects what is already stored, once. It only touches +-- rows that are actually wrong, so it is safe to run again. +UPDATE folders f + SET unread_count = c.unread, + total_count = c.total + FROM ( + SELECT f2.id, + COUNT(e.id) FILTER ( + WHERE NOT EXISTS ( + SELECT 1 FROM message_flags mf + WHERE mf.message_id = e.id + AND mf.received_at = e.received_at + AND mf.flag = '\Seen') + )::int AS unread, + COUNT(e.id)::int AS total + FROM folders f2 + LEFT JOIN email_messages e + ON e.folder_id = f2.id AND e.expunged_at IS NULL + GROUP BY f2.id + ) AS c + WHERE f.id = c.id + AND (f.unread_count IS DISTINCT FROM c.unread + OR f.total_count IS DISTINCT FROM c.total); + +UPDATE mailboxes m + SET used_bytes = COALESCE(s.total, 0) + FROM mailboxes m2 + LEFT JOIN ( + SELECT mailbox_id, SUM(size_bytes)::bigint AS total + FROM email_messages + WHERE expunged_at IS NULL + GROUP BY mailbox_id + ) AS s ON s.mailbox_id = m2.id + WHERE m.id = m2.id + AND m.used_bytes IS DISTINCT FROM COALESCE(s.total, 0); diff --git a/services/auth/src/cli.test.ts b/services/auth/src/cli.test.ts new file mode 100644 index 0000000..f538aa6 --- /dev/null +++ b/services/auth/src/cli.test.ts @@ -0,0 +1,46 @@ +import { describe, it, expect, afterAll } from "vitest"; + +process.env.LAMBDAMAIL_MASTER_KEY ||= "test-master-key-at-least-16-chars"; +process.env.DATABASE_URL = process.env.TEST_DATABASE_URL ?? ""; + +const hasDatabase = Boolean(process.env.TEST_DATABASE_URL); +const describeDb = hasDatabase ? describe : describe.skip; + +const { createAdmin, CliError } = await import("./cli.js"); +const { query, closePool } = await import("./db.js"); + +const DOMAIN = `cli-${Date.now()}.test`; +const EMAIL = `root@${DOMAIN}`; + +describeDb("create-admin (command line)", () => { + afterAll(async () => { + await query(`DELETE FROM domains WHERE name = $1`, [DOMAIN]); + await closePool(); + }); + + it("gives the first administrator every standard folder, Reports included", async () => { + const log = console.log; + console.log = () => undefined; + try { + await createAdmin(EMAIL, "a-long-enough-password", true); + } finally { + console.log = log; + } + + // This account used to come out without Reports, so its DMARC and TLS-RPT + // reports fell into the INBOX: the console and this command each had their + // own list of folders and the lists had drifted. + const folders = await query<{ name: string; special_use: string | null }>( + `SELECT f.name, f.special_use FROM folders f JOIN mailboxes m ON m.id = f.mailbox_id WHERE m.email_address = $1`, + [EMAIL], + ); + expect(folders.map((f) => f.name).sort()).toEqual(["Archive", "Drafts", "INBOX", "Junk", "Reports", "Sent", "Trash"]); + expect(folders.find((f) => f.name === "Reports")?.special_use).toBeNull(); + expect(folders.find((f) => f.name === "Sent")?.special_use).toBe("sent"); + }); + + it("refuses with a readable error instead of exiting the process", async () => { + await expect(createAdmin(EMAIL, "short", true)).rejects.toBeInstanceOf(CliError); + await expect(createAdmin(EMAIL, "a-long-enough-password", true)).rejects.toThrow(/already exists/); + }); +}); diff --git a/services/auth/src/cli.ts b/services/auth/src/cli.ts index 57fce04..7f3351c 100644 --- a/services/auth/src/cli.ts +++ b/services/auth/src/cli.ts @@ -13,22 +13,19 @@ * node dist/cli.js reset-mfa */ +import { pathToFileURL } from "node:url"; import { query, queryOne, closePool } from "./db.js"; import { hashPassword } from "./crypto.js"; const SYSTEM_ALIASES = ["postmaster", "abuse", "dmarc", "tlsrpt"]; -const STANDARD_FOLDERS: Array<[string, string]> = [ - ["INBOX", "inbox"], - ["Sent", "sent"], - ["Drafts", "drafts"], - ["Trash", "trash"], - ["Junk", "junk"], - ["Archive", "archive"], -]; +/** A refusal the operator should read, as opposed to a bug. */ +export class CliError extends Error {} + +// Thrown rather than exiting here, so the commands can be exercised from a test; +// main() prints it as "error: ..." and exits 1, exactly as this used to. function fail(message: string): never { - console.error(`error: ${message}`); - process.exit(1); + throw new CliError(message); } function parseAddress(email: string): { localPart: string; domain: string } { @@ -37,7 +34,7 @@ function parseAddress(email: string): { localPart: string; domain: string } { return { localPart: email.slice(0, at).toLowerCase(), domain: email.slice(at + 1).toLowerCase() }; } -async function createAdmin(email: string, password: string, force: boolean): Promise { +export async function createAdmin(email: string, password: string, force: boolean): Promise { if (password.length < 12) fail("the password must be at least 12 characters"); const { localPart, domain } = parseAddress(email); @@ -69,13 +66,10 @@ async function createAdmin(email: string, password: string, force: boolean): Pro ); // Delivery needs these: a mailbox with no INBOX cannot receive at all, and - // one with no Junk used to fail the spam path outright. - for (const [name, specialUse] of STANDARD_FOLDERS) { - await query( - `INSERT INTO folders (mailbox_id, name, special_use) VALUES ($1, $2, $3)`, - [mailbox!.id, name, specialUse], - ); - } + // one with no Junk used to fail the spam path outright. The list lives in the + // database (migration 0014) so the console and this command cannot disagree + // about it, as they once did over the Reports folder. + await query(`SELECT provision_default_folders($1)`, [mailbox!.id]); // PLAN.md section 7.4b: the published DNS records name these, and reports // bounce if they do not resolve. @@ -174,10 +168,13 @@ async function main(): Promise { } } -main() - .then(() => closePool()) - .catch(async (err) => { - console.error(`error: ${err instanceof Error ? err.message : String(err)}`); - await closePool().catch(() => undefined); - process.exit(1); - }); +// Only when run as a command, not when a test imports the functions above +if (process.argv[1] && import.meta.url === pathToFileURL(process.argv[1]).href) { + main() + .then(() => closePool()) + .catch(async (err) => { + console.error(`error: ${err instanceof Error ? err.message : String(err)}`); + await closePool().catch(() => undefined); + process.exit(1); + }); +} diff --git a/services/auth/src/repository.ts b/services/auth/src/repository.ts index 45457a2..0560da4 100644 --- a/services/auth/src/repository.ts +++ b/services/auth/src/repository.ts @@ -1004,18 +1004,9 @@ export async function createMailbox(scope: AdminScope, input: CreateMailboxInput [input.domainId, localPart, email, hash, input.role, input.quotaBytes], ); - await query( - `INSERT INTO folders (mailbox_id, name, special_use) - SELECT $1, f.name, f.special_use - FROM (VALUES ('INBOX','inbox'), ('Sent','sent'), ('Drafts','drafts'), - ('Trash','trash'), ('Junk','junk'), ('Archive','archive'), - -- Where delivered DMARC and TLS-RPT reports are filed. - -- IMAP has no special-use role for these, so it is matched - -- by name. Without the folder the delivery path falls back - -- to INBOX, which is the noise this is here to remove. - ('Reports', NULL)) AS f(name, special_use)`, - [created!.id], - ); + // The standard folders come from one definition in the database (migration + // 0014), shared with the command that creates the first administrator. + await query(`SELECT provision_default_folders($1)`, [created!.id]); return { status: "OK", id: created!.id, email }; } diff --git a/services/protocols/internal/infrastructure/postgres/mailbox_state_migration_test.go b/services/protocols/internal/infrastructure/postgres/mailbox_state_migration_test.go new file mode 100644 index 0000000..81be194 --- /dev/null +++ b/services/protocols/internal/infrastructure/postgres/mailbox_state_migration_test.go @@ -0,0 +1,158 @@ +package postgres + +import ( + "context" + "os" + "testing" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgxpool" +) + +// runMigration0014 applies the migration file itself, so the test cannot drift +// from what production runs. It is idempotent by design, which is also tested. +func runMigration0014(t *testing.T, pool *pgxpool.Pool) { + t.Helper() + sql, err := os.ReadFile("../../../../../migrations/0014_mailbox_state_single_source.up.sql") + if err != nil { + t.Fatalf("read migration: %v", err) + } + if _, err := pool.Exec(context.Background(), string(sql)); err != nil { + t.Fatalf("apply migration 0014: %v", err) + } +} + +func folderNames(t *testing.T, pool *pgxpool.Pool, mailboxID string) map[string]string { + t.Helper() + rows, err := pool.Query(context.Background(), `SELECT name, COALESCE(special_use, '') FROM folders WHERE mailbox_id = $1`, mailboxID) + if err != nil { + t.Fatalf("list folders: %v", err) + } + defer rows.Close() + out := map[string]string{} + for rows.Next() { + var name, role string + if err := rows.Scan(&name, &role); err != nil { + t.Fatalf("scan: %v", err) + } + out[name] = role + } + return out +} + +// Numbers that two paths left wrong are corrected from the rows. +func TestMigration0014_ReconcilesDriftedCountersAndQuota(t *testing.T) { + pool := testPool(t) + ctx := context.Background() + mailboxID := seedMailbox(t, pool) + deliver(t, pool, mailboxID, "INBOX", "a", false) + deliver(t, pool, mailboxID, "INBOX", "b", true) + deliver(t, pool, mailboxID, "Sent", "c", true) + + // The drift as it was found: counters and quota that disagree with the rows + if _, err := pool.Exec(ctx, `UPDATE folders SET unread_count = 9, total_count = 7 WHERE mailbox_id = $1`, mailboxID); err != nil { + t.Fatalf("corrupt counters: %v", err) + } + if _, err := pool.Exec(ctx, `UPDATE mailboxes SET used_bytes = 12345 WHERE id = $1`, mailboxID); err != nil { + t.Fatalf("corrupt quota: %v", err) + } + + runMigration0014(t, pool) + assertConsistent(t, pool, mailboxID, "the reconciliation") + + runMigration0014(t, pool) + assertConsistent(t, pool, mailboxID, "running the reconciliation a second time") +} + +// A mailbox that is already right is left exactly as it was. +func TestMigration0014_LeavesATrueMailboxAlone(t *testing.T) { + pool := testPool(t) + ctx := context.Background() + mailboxID := seedMailbox(t, pool) + deliver(t, pool, mailboxID, "INBOX", "a", false) + before, usedBefore, _ := mailState(t, pool, mailboxID) + + runMigration0014(t, pool) + + after, usedAfter, _ := mailState(t, pool, mailboxID) + for name, f := range before { + if after[name].CachedTotal != f.CachedTotal || after[name].CachedUnread != f.CachedUnread || after[name].HighestModSeq != f.HighestModSeq { + t.Errorf("%s changed: %+v -> %+v", name, f, after[name]) + } + } + if usedAfter != usedBefore { + t.Errorf("used_bytes changed from %d to %d", usedBefore, usedAfter) + } + _ = ctx +} + +// The one list of standard folders: what is missing is added, what is there is +// not duplicated, and running it again changes nothing. +func TestProvisionDefaultFolders(t *testing.T) { + pool := testPool(t) + ctx := context.Background() + runMigration0014(t, pool) + + domainID, mailboxID := uuid.New(), uuid.New() + name := "prov" + uuid.NewString()[:8] + ".test" + if _, err := pool.Exec(ctx, `INSERT INTO domains (id, name, punycode_name, is_active) VALUES ($1,$2,$3,true)`, domainID, name, name); err != nil { + t.Fatalf("seed domain: %v", err) + } + t.Cleanup(func() { _, _ = pool.Exec(context.Background(), `DELETE FROM domains WHERE id = $1`, domainID) }) + if _, err := pool.Exec(ctx, `INSERT INTO mailboxes (id, domain_id, local_part, email_address, password_hash, is_active) VALUES ($1,$2,'u',$3,'x',true)`, + mailboxID, domainID, "u@"+name); err != nil { + t.Fatalf("seed mailbox: %v", err) + } + + want := map[string]string{"INBOX": "inbox", "Sent": "sent", "Drafts": "drafts", "Trash": "trash", "Junk": "junk", "Archive": "archive", "Reports": ""} + + if _, err := pool.Exec(ctx, `SELECT provision_default_folders($1)`, mailboxID); err != nil { + t.Fatalf("provision: %v", err) + } + got := folderNames(t, pool, mailboxID.String()) + for n, role := range want { + if r, ok := got[n]; !ok || r != role { + t.Errorf("folder %q: present=%v role=%q, want role %q", n, ok, r, role) + } + } + if len(got) != len(want) { + t.Errorf("got %d folders, want %d: %v", len(got), len(want), got) + } + + if _, err := pool.Exec(ctx, `SELECT provision_default_folders($1)`, mailboxID); err != nil { + t.Fatalf("provision again: %v", err) + } + if again := folderNames(t, pool, mailboxID.String()); len(again) != len(want) { + t.Errorf("a second call changed the folder count to %d", len(again)) + } +} + +// A folder that exists under another capitalisation counts as present, and the +// folder keeps the owner's mail. +func TestProvisionDefaultFolders_DoesNotDuplicateByCase(t *testing.T) { + pool := testPool(t) + ctx := context.Background() + runMigration0014(t, pool) + mailboxID := seedMailbox(t, pool) // INBOX, Sent, Drafts, Trash + + if _, err := pool.Exec(ctx, `INSERT INTO folders (mailbox_id, name, special_use) VALUES ($1, 'reports', NULL)`, mailboxID); err != nil { + t.Fatalf("seed: %v", err) + } + if _, err := pool.Exec(ctx, `SELECT provision_default_folders($1)`, mailboxID); err != nil { + t.Fatalf("provision: %v", err) + } + + got := folderNames(t, pool, mailboxID) + reports := 0 + for n := range got { + if n == "reports" || n == "Reports" { + reports++ + } + } + if reports != 1 { + t.Errorf("found %d reports folders, want 1: %v", reports, got) + } + if len(got) != 7 { + t.Errorf("got %d folders, want 7: %v", len(got), got) + } +} From 67bbe7a50a06caa7ce8f4f60111eec238eba355e Mon Sep 17 00:00:00 2001 From: Lucas Rayan Date: Fri, 16 Oct 2026 13:22:05 -0300 Subject: [PATCH 4/6] refactor(mail): IMAP and the webmail read folders and messages through one MailStore The folder list and the message list were two queries over the same tables. IMAP took a folder's message count from a cached column and the webmail counted rows, so the two could disagree. Both now ask MailStore, which counts the messages. --- .../postgres/imap_folder_repository.go | 53 ++--- .../postgres/mail_store_read.go | 202 ++++++++++++++++++ .../postgres/message_query_repository.go | 31 +-- .../postgres/webmail_repository.go | 156 +++++--------- 4 files changed, 287 insertions(+), 155 deletions(-) create mode 100644 services/protocols/internal/infrastructure/postgres/mail_store_read.go diff --git a/services/protocols/internal/infrastructure/postgres/imap_folder_repository.go b/services/protocols/internal/infrastructure/postgres/imap_folder_repository.go index ee8a6b9..12ecd3a 100644 --- a/services/protocols/internal/infrastructure/postgres/imap_folder_repository.go +++ b/services/protocols/internal/infrastructure/postgres/imap_folder_repository.go @@ -3,57 +3,50 @@ package postgres import ( "context" - "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" "lambdamail/protocols/internal/application/port" ) -// ImapFolderRepository implements port.ImapFolderRepository against Postgres. +// ImapFolderRepository resolves a mailbox's folders for LIST and SELECT. The +// question is answered by MailStore, the same one the webmail asks, so a folder +// reports the same number of messages on both. type ImapFolderRepository struct { - pool *pgxpool.Pool + store *MailStore } func NewImapFolderRepository(pool *pgxpool.Pool) *ImapFolderRepository { - return &ImapFolderRepository{pool: pool} + return &ImapFolderRepository{store: NewMailStore(pool)} +} + +func imapRecord(f FolderInfo) port.ImapFolderRecord { + return port.ImapFolderRecord{ + ID: f.ID, + Name: f.Name, + UIDNext: uint32(f.UIDNext), + UIDValidity: uint32(f.UIDValidity), + NumMessages: uint32(f.Total), + HighestModSeq: uint64(f.HighestModSeq), + } } func (r *ImapFolderRepository) FindByName(ctx context.Context, mailboxID, name string) (*port.ImapFolderRecord, error) { - row := r.pool.QueryRow(ctx, ` - SELECT f.id, f.name, f.uid_next, f.uid_validity, f.total_count, f.highest_modseq - FROM folders f - WHERE f.mailbox_id = $1 AND f.name = $2 - `, mailboxID, name) - - var rec port.ImapFolderRecord - if err := row.Scan(&rec.ID, &rec.Name, &rec.UIDNext, &rec.UIDValidity, &rec.NumMessages, &rec.HighestModSeq); err != nil { - if err == pgx.ErrNoRows { - return nil, nil - } + f, err := r.store.FolderByName(ctx, mailboxID, name) + if err != nil || f == nil { return nil, err } + rec := imapRecord(*f) return &rec, nil } func (r *ImapFolderRepository) ListFolders(ctx context.Context, mailboxID string) ([]port.ImapFolderRecord, error) { - rows, err := r.pool.Query(ctx, ` - SELECT f.id, f.name, f.uid_next, f.uid_validity, f.total_count, f.highest_modseq - FROM folders f - WHERE f.mailbox_id = $1 - ORDER BY f.name - `, mailboxID) + folders, err := r.store.Folders(ctx, mailboxID) if err != nil { return nil, err } - defer rows.Close() - var out []port.ImapFolderRecord - for rows.Next() { - var rec port.ImapFolderRecord - if err := rows.Scan(&rec.ID, &rec.Name, &rec.UIDNext, &rec.UIDValidity, &rec.NumMessages, &rec.HighestModSeq); err != nil { - return nil, err - } - out = append(out, rec) + for _, f := range folders { + out = append(out, imapRecord(f)) } - return out, rows.Err() + return out, nil } diff --git a/services/protocols/internal/infrastructure/postgres/mail_store_read.go b/services/protocols/internal/infrastructure/postgres/mail_store_read.go new file mode 100644 index 0000000..7b53943 --- /dev/null +++ b/services/protocols/internal/infrastructure/postgres/mail_store_read.go @@ -0,0 +1,202 @@ +package postgres + +import ( + "context" + "errors" + "fmt" + "time" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5" +) + +// The read side of MailStore. +// +// IMAP (LIST, SELECT, FETCH) and the webmail (folder list, message list) used to +// read the same tables with different queries, and so could answer differently: +// IMAP took a folder's message count from a cached column, the webmail counted +// the rows. A client comparing the two saw numbers that disagreed. They now ask +// the same question, and the answer is derived from the messages themselves, so +// a folder can only ever report what is actually in it. + +// FolderInfo is one folder with the numbers a client needs. +type FolderInfo struct { + ID string + Name string + SpecialUse string + UIDNext int64 + UIDValidity int64 + HighestModSeq int64 + // Total and Unread are counted from the messages, not read from the cached + // columns: the columns are kept true by every write path, and counting here + // makes a wrong one visible instead of repeated. + Total int + Unread int +} + +const folderColumns = ` + SELECT f.id::text, f.name, COALESCE(f.special_use, ''), + f.uid_next, f.uid_validity, f.highest_modseq, + COUNT(e.id)::int, + COUNT(e.id) FILTER ( + WHERE NOT EXISTS (SELECT 1 FROM message_flags mf + WHERE mf.message_id = e.id + AND mf.received_at = e.received_at + AND mf.flag = '\Seen') + )::int + FROM folders f + LEFT JOIN email_messages e ON e.folder_id = f.id AND e.expunged_at IS NULL` + +func scanFolder(row pgx.Row) (FolderInfo, error) { + var f FolderInfo + err := row.Scan(&f.ID, &f.Name, &f.SpecialUse, &f.UIDNext, &f.UIDValidity, &f.HighestModSeq, &f.Total, &f.Unread) + return f, err +} + +// Folders lists a mailbox's folders by name. +func (s *MailStore) Folders(ctx context.Context, mailboxID string) ([]FolderInfo, error) { + rows, err := s.pool.Query(ctx, folderColumns+` + WHERE f.mailbox_id = $1 + GROUP BY f.id + ORDER BY f.name`, mailboxID) + if err != nil { + return nil, fmt.Errorf("list folders: %w", err) + } + defer rows.Close() + + var out []FolderInfo + for rows.Next() { + f, err := scanFolder(rows) + if err != nil { + return nil, err + } + out = append(out, f) + } + return out, rows.Err() +} + +// FolderByName finds one folder by its exact name; nil, nil when there is none +// (IMAP SELECT treats that as a normal answer, not a failure). +func (s *MailStore) FolderByName(ctx context.Context, mailboxID, name string) (*FolderInfo, error) { + f, err := scanFolder(s.pool.QueryRow(ctx, folderColumns+` + WHERE f.mailbox_id = $1 AND f.name = $2 + GROUP BY f.id`, mailboxID, name)) + if errors.Is(err, pgx.ErrNoRows) { + return nil, nil + } + if err != nil { + return nil, err + } + return &f, nil +} + +// MessageInfo is one message as either a mail client or the webmail lists it. +type MessageInfo struct { + UID uint32 + BlobID uuid.UUID + Size int64 + ReceivedAt time.Time + ModSeq uint64 + Flags []string + Subject string + SenderAddress string + FromName string + Snippet string + HasAttachments bool + SpamVerdict string + DmarcResult string +} + +// HasFlag reports whether the message carries the flag. +func (m MessageInfo) HasFlag(flag string) bool { + for _, f := range m.Flags { + if f == flag { + return true + } + } + return false +} + +// MessageQuery shapes a listing. The zero value lists every message of the +// folder by UID ascending, which is what IMAP needs to number them. +type MessageQuery struct { + // NewestFirst orders by arrival, newest first, as the webmail shows them + NewestFirst bool + // Search keeps messages whose subject, sender or snippet contains it + Search string + // Limit and Offset page the result; Limit 0 means no limit + Limit int + Offset int +} + +// Messages lists a folder's live messages with their flags. +func (s *MailStore) Messages(ctx context.Context, folderID string, q MessageQuery) ([]MessageInfo, error) { + order := `e.uid ASC` + if q.NewestFirst { + order = `e.received_at DESC, e.uid DESC` + } + limit := q.Limit + if limit <= 0 { + limit = 1<<31 - 1 + } + offset := q.Offset + if offset < 0 { + offset = 0 + } + + rows, err := s.pool.Query(ctx, ` + SELECT e.uid, e.blob_id, e.size_bytes, e.received_at, e.modseq, + COALESCE((SELECT array_agg(mf.flag) FROM message_flags mf + WHERE mf.message_id = e.id AND mf.received_at = e.received_at), '{}'), + COALESCE(e.subject, ''), e.sender_address::text, + COALESCE(e.from_display_name, ''), COALESCE(e.snippet, ''), + e.has_attachments, COALESCE(e.spam_verdict, ''), COALESCE(e.dmarc_result, '') + FROM email_messages e + WHERE e.folder_id = $1 AND e.expunged_at IS NULL + AND ($2 = '' OR e.subject ILIKE '%' || $2 || '%' + OR e.sender_address::text ILIKE '%' || $2 || '%' + OR e.snippet ILIKE '%' || $2 || '%') + ORDER BY `+order+` + LIMIT $3 OFFSET $4`, folderID, q.Search, limit, offset) + if err != nil { + return nil, fmt.Errorf("list messages: %w", err) + } + defer rows.Close() + + var out []MessageInfo + for rows.Next() { + var m MessageInfo + if err := rows.Scan(&m.UID, &m.BlobID, &m.Size, &m.ReceivedAt, &m.ModSeq, &m.Flags, + &m.Subject, &m.SenderAddress, &m.FromName, &m.Snippet, + &m.HasAttachments, &m.SpamVerdict, &m.DmarcResult); err != nil { + return nil, err + } + out = append(out, m) + } + return out, rows.Err() +} + +// MessageBlob resolves one message to its stored blob, or ErrMessageNotFound. +func (s *MailStore) MessageBlob(ctx context.Context, folderID string, uid uint32) (uuid.UUID, error) { + var blobID uuid.UUID + err := s.pool.QueryRow(ctx, ` + SELECT blob_id FROM email_messages WHERE folder_id = $1 AND uid = $2 AND expunged_at IS NULL`, + folderID, int64(uid)).Scan(&blobID) + if errors.Is(err, pgx.ErrNoRows) { + return uuid.Nil, ErrMessageNotFound + } + return blobID, err +} + +// MailboxIDByAddress resolves an active address to its mailbox id. +func (s *MailStore) MailboxIDByAddress(ctx context.Context, address string) (string, error) { + var id string + err := s.pool.QueryRow(ctx, ` + SELECT m.id::text FROM mailboxes m + JOIN domains d ON d.id = m.domain_id + WHERE m.email_address = $1 AND m.is_active AND d.is_active`, address).Scan(&id) + if errors.Is(err, pgx.ErrNoRows) { + return "", ErrMessageNotFound + } + return id, err +} diff --git a/services/protocols/internal/infrastructure/postgres/message_query_repository.go b/services/protocols/internal/infrastructure/postgres/message_query_repository.go index 09a0c0c..9f647ef 100644 --- a/services/protocols/internal/infrastructure/postgres/message_query_repository.go +++ b/services/protocols/internal/infrastructure/postgres/message_query_repository.go @@ -8,37 +8,26 @@ import ( "lambdamail/protocols/internal/application/port" ) -// MessageQueryRepository implements port.MessageQueryRepository against Postgres. +// MessageQueryRepository lists a folder's messages for IMAP FETCH, through the +// same MailStore listing the webmail uses. type MessageQueryRepository struct { - pool *pgxpool.Pool + store *MailStore } func NewMessageQueryRepository(pool *pgxpool.Pool) *MessageQueryRepository { - return &MessageQueryRepository{pool: pool} + return &MessageQueryRepository{store: NewMailStore(pool)} } func (r *MessageQueryRepository) ListMessages(ctx context.Context, folderID string) ([]port.MessageRecord, error) { - rows, err := r.pool.Query(ctx, ` - SELECT m.uid, m.blob_id, m.size_bytes, m.received_at, m.modseq, - COALESCE(array_agg(f.flag) FILTER (WHERE f.flag IS NOT NULL), '{}') - FROM email_messages m - LEFT JOIN message_flags f ON f.message_id = m.id AND f.received_at = m.received_at - WHERE m.folder_id = $1 AND m.expunged_at IS NULL - GROUP BY m.uid, m.blob_id, m.size_bytes, m.received_at, m.modseq - ORDER BY m.uid ASC - `, folderID) + messages, err := r.store.Messages(ctx, folderID, MessageQuery{}) if err != nil { return nil, err } - defer rows.Close() - var out []port.MessageRecord - for rows.Next() { - var rec port.MessageRecord - if err := rows.Scan(&rec.UID, &rec.BlobID, &rec.SizeBytes, &rec.ReceivedAt, &rec.ModSeq, &rec.Flags); err != nil { - return nil, err - } - out = append(out, rec) + for _, m := range messages { + out = append(out, port.MessageRecord{ + UID: m.UID, BlobID: m.BlobID, SizeBytes: m.Size, ReceivedAt: m.ReceivedAt, Flags: m.Flags, ModSeq: m.ModSeq, + }) } - return out, rows.Err() + return out, nil } diff --git a/services/protocols/internal/infrastructure/postgres/webmail_repository.go b/services/protocols/internal/infrastructure/postgres/webmail_repository.go index e4adef9..bd1521f 100644 --- a/services/protocols/internal/infrastructure/postgres/webmail_repository.go +++ b/services/protocols/internal/infrastructure/postgres/webmail_repository.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "sort" "github.com/google/uuid" "github.com/jackc/pgx/v5" @@ -27,135 +28,82 @@ func NewWebmailRepository(pool *pgxpool.Pool) *WebmailRepository { return &WebmailRepository{pool: pool, store: NewMailStore(pool)} } +// The methods below only read. They do not query on their own: the webmail asks +// MailStore the same questions IMAP does and shapes the answers for the +// browser, so a folder shows the same numbers in both. + func (r *WebmailRepository) FindMailboxIDByAddress(ctx context.Context, address string) (string, error) { - var id string - err := r.pool.QueryRow(ctx, ` - SELECT m.id::text FROM mailboxes m - JOIN domains d ON d.id = m.domain_id - WHERE m.email_address = $1 AND m.is_active AND d.is_active - `, address).Scan(&id) - if errors.Is(err, pgx.ErrNoRows) { - return "", ErrMessageNotFound - } - return id, err + return r.store.MailboxIDByAddress(ctx, address) } +// roleOrder is how the folder list is arranged for people: the folders they use +// every day first, in a fixed order, then their own by name. +var roleOrder = map[string]int{"inbox": 0, "drafts": 1, "sent": 2, "archive": 3, "junk": 4, "trash": 5} + func (r *WebmailRepository) ListFolders(ctx context.Context, mailboxID string) ([]port.WebmailFolder, error) { - // The counts are derived from the messages themselves rather than read out - // of folders.unread_count/total_count. - // - // Those columns are a denormalised cache maintained by every write path, - // and once any one of them was wrong nothing brought it back: the number - // the user saw stayed wrong through a reload, a logout and a fresh login, - // because the wrongness was stored. Deriving here is authoritative and - // self-healing - a folder can only ever report what is actually in it. - // - // The columns stay maintained for IMAP's sake (STATUS answers from them), - // so this is a correction at the read side, not their removal. - rows, err := r.pool.Query(ctx, ` - SELECT f.id::text, f.name, COALESCE(f.special_use, ''), - COUNT(e.id) FILTER ( - WHERE NOT EXISTS (SELECT 1 FROM message_flags mf - WHERE mf.message_id = e.id - AND mf.received_at = e.received_at - AND mf.flag = '\Seen') - )::int AS unread_count, - COUNT(e.id)::int AS total_count - FROM folders f - LEFT JOIN email_messages e - ON e.folder_id = f.id AND e.expunged_at IS NULL - WHERE f.mailbox_id = $1 - GROUP BY f.id, f.name, f.special_use - ORDER BY CASE f.special_use - WHEN 'inbox' THEN 0 WHEN 'drafts' THEN 1 WHEN 'sent' THEN 2 - WHEN 'archive' THEN 3 WHEN 'junk' THEN 4 WHEN 'trash' THEN 5 - ELSE 6 END, f.name - `, mailboxID) + folders, err := r.store.Folders(ctx, mailboxID) if err != nil { - return nil, fmt.Errorf("list folders: %w", err) + return nil, err } - defer rows.Close() - folders := []port.WebmailFolder{} - for rows.Next() { - var f port.WebmailFolder - if err := rows.Scan(&f.ID, &f.Name, &f.SpecialUse, &f.UnreadCount, &f.TotalCount); err != nil { - return nil, err + out := make([]port.WebmailFolder, 0, len(folders)) + for _, f := range folders { + out = append(out, port.WebmailFolder{ID: f.ID, Name: f.Name, SpecialUse: f.SpecialUse, UnreadCount: f.Unread, TotalCount: f.Total}) + } + + rank := func(f port.WebmailFolder) int { + if n, ok := roleOrder[f.SpecialUse]; ok { + return n } - folders = append(folders, f) + return 6 } - return folders, rows.Err() + sort.SliceStable(out, func(i, j int) bool { + if ri, rj := rank(out[i]), rank(out[j]); ri != rj { + return ri < rj + } + return out[i].Name < out[j].Name + }) + return out, nil } -// folderCondition matches either the special-use role ("inbox") or the literal -// folder name, so the UI can address a folder the same way IMAP does. -const folderCondition = `(f.special_use = LOWER($2) OR LOWER(f.name) = LOWER($2))` - func (r *WebmailRepository) ListMessages( ctx context.Context, mailboxID, folderName, search string, limit, offset int, ) ([]port.WebmailMessage, error) { if limit <= 0 || limit > 200 { limit = 50 } - if offset < 0 { - offset = 0 - } - rows, err := r.pool.Query(ctx, ` - SELECT e.uid, COALESCE(e.subject, ''), e.sender_address::text, - COALESCE(e.from_display_name, ''), COALESCE(e.snippet, ''), - e.size_bytes, e.received_at, e.has_attachments, - EXISTS (SELECT 1 FROM message_flags mf - WHERE mf.message_id = e.id AND mf.received_at = e.received_at AND mf.flag = '\Seen'), - EXISTS (SELECT 1 FROM message_flags mf - WHERE mf.message_id = e.id AND mf.received_at = e.received_at AND mf.flag = '\Flagged'), - EXISTS (SELECT 1 FROM message_flags mf - WHERE mf.message_id = e.id AND mf.received_at = e.received_at AND mf.flag = '\Answered'), - COALESCE(e.spam_verdict, ''), COALESCE(e.dmarc_result, '') - FROM email_messages e - JOIN folders f ON f.id = e.folder_id - WHERE e.mailbox_id = $1 AND `+folderCondition+` - AND e.expunged_at IS NULL - AND ($3 = '' OR e.subject ILIKE '%%' || $3 || '%%' - OR e.sender_address::text ILIKE '%%' || $3 || '%%' - OR e.snippet ILIKE '%%' || $3 || '%%') - ORDER BY e.received_at DESC, e.uid DESC - LIMIT $4 OFFSET $5 - `, mailboxID, folderName, search, limit, offset) + folderID, err := r.store.ResolveFolder(ctx, mailboxID, folderName) + if errors.Is(err, port.ErrFolderMissing) { + return []port.WebmailMessage{}, nil + } if err != nil { - return nil, fmt.Errorf("list messages: %w", err) + return nil, err } - defer rows.Close() - messages := []port.WebmailMessage{} - for rows.Next() { - var m port.WebmailMessage - if err := rows.Scan( - &m.UID, &m.Subject, &m.SenderAddress, &m.FromName, &m.Snippet, - &m.SizeBytes, &m.ReceivedAt, &m.HasAttachments, - &m.Seen, &m.Flagged, &m.Answered, &m.SpamVerdict, &m.DmarcResult, - ); err != nil { - return nil, err - } - messages = append(messages, m) + messages, err := r.store.Messages(ctx, folderID, MessageQuery{NewestFirst: true, Search: search, Limit: limit, Offset: offset}) + if err != nil { + return nil, err } - return messages, rows.Err() + + out := make([]port.WebmailMessage, 0, len(messages)) + for _, m := range messages { + out = append(out, port.WebmailMessage{ + UID: m.UID, Subject: m.Subject, SenderAddress: m.SenderAddress, FromName: m.FromName, Snippet: m.Snippet, + SizeBytes: m.Size, ReceivedAt: m.ReceivedAt, HasAttachments: m.HasAttachments, + Seen: m.HasFlag(`\Seen`), Flagged: m.HasFlag(`\Flagged`), Answered: m.HasFlag(`\Answered`), + SpamVerdict: m.SpamVerdict, DmarcResult: m.DmarcResult, + }) + } + return out, nil } -func (r *WebmailRepository) GetMessageBlob( - ctx context.Context, mailboxID, folderName string, uid uint32, -) (uuid.UUID, error) { - var blobID uuid.UUID - err := r.pool.QueryRow(ctx, ` - SELECT e.blob_id FROM email_messages e - JOIN folders f ON f.id = e.folder_id - WHERE e.mailbox_id = $1 AND `+folderCondition+` - AND e.uid = $3 AND e.expunged_at IS NULL - `, mailboxID, folderName, int64(uid)).Scan(&blobID) - if errors.Is(err, pgx.ErrNoRows) { - return uuid.Nil, ErrMessageNotFound +func (r *WebmailRepository) GetMessageBlob(ctx context.Context, mailboxID, folderName string, uid uint32) (uuid.UUID, error) { + folderID, err := r.messageFolder(ctx, mailboxID, folderName) + if err != nil { + return uuid.Nil, err } - return blobID, err + return r.store.MessageBlob(ctx, folderID, uid) } // The methods below change state. They do not do it themselves: the webmail From 183191d18d50ca4b1e492f78e57de15467cdcdf7 Mon Sep 17 00:00:00 2001 From: Lucas Rayan Date: Fri, 16 Oct 2026 13:30:59 -0300 Subject: [PATCH 5/6] refactor(mail): one send path for Sent, a fixed lock order, and concurrency tests - the webmail's Send no longer files its own Sent copy: it asks Submit to (FileSentCopy), the same path SMTP clients use, so the rule exists once - every operation now locks the folders first (lowest id first) and the message after. Opposite moves, and a move racing a flag change, deadlocked (SQLSTATE 40P01) because they took those locks in different orders; the new test reproduces it (10-19 s of retries, errors) and passes in ~1 s with the order - TestOneMailboxEverywhere drives SMTP, IMAP and the webmail on one mailbox, mixing them, and checks after each step that every view agrees with the others and with the rows. It fails on the previous code and passes now. --- .../application/usecase/sent_copy_test.go | 48 +++ .../usecase/webmail_compose_test.go | 2 + .../application/usecase/webmail_mail.go | 53 +-- .../e2e/one_mailbox_everywhere_e2e_test.go | 385 ++++++++++++++++++ .../postgres/mail_state_contract_test.go | 129 ++++++ .../infrastructure/postgres/mail_store.go | 43 +- 6 files changed, 612 insertions(+), 48 deletions(-) create mode 100644 services/protocols/internal/e2e/one_mailbox_everywhere_e2e_test.go diff --git a/services/protocols/internal/application/usecase/sent_copy_test.go b/services/protocols/internal/application/usecase/sent_copy_test.go index 019783d..52c5146 100644 --- a/services/protocols/internal/application/usecase/sent_copy_test.go +++ b/services/protocols/internal/application/usecase/sent_copy_test.go @@ -137,3 +137,51 @@ func TestSubmit_WorksWithoutASentCopyRepository(t *testing.T) { t.Fatalf("Submit without a Sent repository: %v", err) } } + +// One rule for every path: a message sent from the webmail and the same message +// sent by an SMTP client leave the same thing behind in Sent. They used to differ +// completely - the webmail filed a copy, SMTP filed nothing - because the rule +// lived in the webmail alone. +func TestSentCopyIsTheSameWhicheverPathSent(t *testing.T) { + mailboxID := uuid.New() + messages := &recordingMessageRepository{} + submission := NewProcessOutboundEmailUseCase(nil, &countingOutboundRepo{}, &capturingBlobStorage{}, nil) + submission.SetSentCopies(messages) + + account := &port.MailboxAuth{ID: mailboxID, EmailAddress: "me@example.test", DomainName: "example.test", MaxRecipientsPerHour: 100} + web := &WebmailUseCase{ + repo: &stubWebmailRepo{mailboxID: mailboxID.String()}, + auth: &stubAuthRepo{account: account}, + submission: submission, + localHost: "mail.example.test", + } + + if err := web.Send(context.Background(), ComposeInput{From: "me@example.test", To: []string{"you@remote.test"}, Subject: "Aviso", Text: "hi"}); err != nil { + t.Fatalf("webmail send: %v", err) + } + + smtp := ProcessOutboundEmailInput{ + MailboxID: mailboxID, SenderAddr: "me@example.test", DomainName: "example.test", + Recipients: []string{"you@remote.test"}, FileSentCopy: true, + Body: bytes.NewBufferString("From: me@example.test\r\nTo: you@remote.test\r\nSubject: Aviso\r\n\r\nhi"), + } + if err := submission.Submit(context.Background(), smtp); err != nil { + t.Fatalf("smtp submit: %v", err) + } + + if len(messages.persisted) != 2 { + t.Fatalf("want one copy per path (2), got %d: the webmail must not file its own copy as well", len(messages.persisted)) + } + fromWeb, fromSMTP := messages.persisted[0], messages.persisted[1] + for _, got := range []port.PersistInboundMessageInput{fromWeb, fromSMTP} { + if got.TargetFolderName != "Sent" || !got.AlreadySeen || got.MailboxID != mailboxID { + t.Errorf("a Sent copy is not what the rule says: %+v", got) + } + if got.SenderAddress != "me@example.test" || got.RecipientAddress != "me@example.test" { + t.Errorf("a Sent copy belongs to its sender: %+v", got) + } + if got.Subject != "Aviso" { + t.Errorf("subject = %q, want Aviso", got.Subject) + } + } +} diff --git a/services/protocols/internal/application/usecase/webmail_compose_test.go b/services/protocols/internal/application/usecase/webmail_compose_test.go index 161abf0..0b7dffd 100644 --- a/services/protocols/internal/application/usecase/webmail_compose_test.go +++ b/services/protocols/internal/application/usecase/webmail_compose_test.go @@ -197,6 +197,8 @@ func newSendableUseCaseWith( } submission := NewProcessOutboundEmailUseCase( nil, &countingOutboundRepo{}, &capturingBlobStorage{}, &recordingSigner{}) + // The Sent copy is filed by the submission, as it is for SMTP clients + submission.SetSentCopies(messages) return (&WebmailUseCase{ repo: repo, auth: &stubAuthRepo{account: account}, diff --git a/services/protocols/internal/application/usecase/webmail_mail.go b/services/protocols/internal/application/usecase/webmail_mail.go index 361c9f2..f67c0af 100644 --- a/services/protocols/internal/application/usecase/webmail_mail.go +++ b/services/protocols/internal/application/usecase/webmail_mail.go @@ -32,10 +32,9 @@ type WebmailUseCase struct { submission *ProcessOutboundEmailUseCase auth port.AuthRepository localHost string - // blobStore and messages file the sender's own copy into Sent, and hold - // drafts. Both are optional: when either is nil the copy is skipped rather - // than the send failing, because a message that has already been queued - // for delivery must not be reported as an error. + // blobStore and messages hold drafts. Both are optional: when either is nil + // saving a draft reports ErrDraftsUnavailable. (The copy in Sent is filed by + // the submission use case, not here.) blobStore port.BlobStorage messages port.InboundMessageRepository } @@ -53,7 +52,7 @@ func NewWebmailUseCase( return &WebmailUseCase{repo: repo, blobs: blobs, submission: submission, auth: auth, localHost: localHost} } -// WithLocalFiling enables the Sent copy and draft storage. +// WithLocalFiling enables draft storage. func (uc *WebmailUseCase) WithLocalFiling(store port.BlobStorage, messages port.InboundMessageRepository) *WebmailUseCase { uc.blobStore = store uc.messages = messages @@ -188,6 +187,10 @@ func (uc *WebmailUseCase) Send(ctx context.Context, input ComposeInput) error { payload := buildMimeMessage(input, recipients, uc.localHost) + // The copy in Sent is filed by Submit, the same way it is for a client that + // sends over SMTP: one rule, one place. It used to be filed here, after the + // submission, and not at all for SMTP, so an account that sent by SMTP had an + // empty Sent folder. if err := uc.submission.Submit(ctx, ProcessOutboundEmailInput{ MailboxID: account.ID, SenderAddr: account.EmailAddress, @@ -195,19 +198,11 @@ func (uc *WebmailUseCase) Send(ctx context.Context, input ComposeInput) error { Recipients: recipients, Body: bytes.NewReader(payload), MaxRecipientsPerHour: account.MaxRecipientsPerHour, + FileSentCopy: true, }); err != nil { return err } - // Only after the message is durably queued. Filing it first would leave a - // copy in Sent for a message that was then rejected by the send limit. - // - // A failure here is logged and swallowed: the mail is already on its way, - // and telling the composer the send failed would invite a duplicate. - if err := uc.fileLocalCopy(ctx, account.ID, account.EmailAddress, "Sent", payload); err != nil { - log.Printf("webmail: could not file the Sent copy for %s: %v", account.EmailAddress, err) - } - // The draft this message grew out of has been superseded by the real // thing. Swallowed like the Sent copy above and for the same reason: the // mail is already queued, and failing the request now would only invite a @@ -288,36 +283,6 @@ func (uc *WebmailUseCase) Delete(ctx context.Context, address, folder string, ui } } -// fileLocalCopy stores a message the user themselves composed into one of -// their own folders. It reuses the inbound persistence path so the copy gets a -// real IMAP UID and shows up over IMAP as well as in the webmail. -func (uc *WebmailUseCase) fileLocalCopy( - ctx context.Context, mailboxID uuid.UUID, address, folder string, payload []byte, -) error { - if uc.blobStore == nil || uc.messages == nil { - return nil - } - blob, err := uc.blobStore.Store(ctx, bytes.NewReader(payload)) - if err != nil { - return err - } - headers := ExtractMessageHeaders(payload) - _, err = uc.messages.PersistAll(ctx, []port.PersistInboundMessageInput{{ - MailboxID: mailboxID, - Blob: blob, - SenderAddress: address, - RecipientAddress: address, - TargetFolderName: folder, - Subject: headers.Subject, - Snippet: headers.Snippet, - FromDisplayName: headers.FromDisplayName, - MessageIDHeader: headers.MessageID, - HasAttachments: headers.HasAttachments, - AlreadySeen: true, - }}) - return err -} - // SaveDraft files the message being composed into the Drafts folder. // // Nothing did this before: the composer displayed "draft saved automatically" diff --git a/services/protocols/internal/e2e/one_mailbox_everywhere_e2e_test.go b/services/protocols/internal/e2e/one_mailbox_everywhere_e2e_test.go new file mode 100644 index 0000000..3153434 --- /dev/null +++ b/services/protocols/internal/e2e/one_mailbox_everywhere_e2e_test.go @@ -0,0 +1,385 @@ +package e2e + +import ( + "bytes" + "context" + "crypto/tls" + "encoding/json" + "fmt" + "net" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/alexedwards/argon2id" + "github.com/emersion/go-imap/v2" + "github.com/emersion/go-imap/v2/imapclient" + "github.com/emersion/go-imap/v2/imapserver" + gosmtp "github.com/emersion/go-smtp" + embeddedpostgres "github.com/fergusstrange/embedded-postgres" + "github.com/google/uuid" + + "lambdamail/protocols/internal/application/port" + appusecase "lambdamail/protocols/internal/application/usecase" + "lambdamail/protocols/internal/infrastructure/diskstorage" + "lambdamail/protocols/internal/infrastructure/dkim" + "lambdamail/protocols/internal/infrastructure/postgres" + httppresentation "lambdamail/protocols/internal/presentation/http" + imappresentation "lambdamail/protocols/internal/presentation/imap" + smtppresentation "lambdamail/protocols/internal/presentation/smtp" +) + +// TestOneMailboxEverywhere is the contract of the whole server: SMTP, IMAP and +// the webmail are three ways into the same mailbox, so whatever one does, the +// other two must see, with the same numbers. +// +// The webmail used to carry its own copy of the rules and its own queries, and +// the two stacks drifted apart: mail sent over SMTP never reached Sent, reading +// over IMAP never moved the unread badge, deleting in the webmail never gave the +// quota back. This test drives all three on one mailbox, mixing them, and checks +// after every step that each view agrees with the others and with the data. +func TestOneMailboxEverywhere(t *testing.T) { + if testing.Short() { + t.Skip("e2e test skipped in -short mode") + } + ctx := context.Background() + + pg := embeddedpostgres.NewDatabase(embeddedpostgres.DefaultConfig(). + Port(15451). + RuntimePath(t.TempDir()). + StartTimeout(120 * time.Second)) + if err := pg.Start(); err != nil { + t.Fatalf("start embedded postgres: %v", err) + } + defer pg.Stop() + + pool, err := postgres.NewPool(ctx, "postgres://postgres:postgres@localhost:15451/postgres?sslmode=disable") + if err != nil { + t.Fatalf("connect pool: %v", err) + } + defer pool.Close() + applyMigrations(t, ctx, pool) + + // ------------------------------------------------------------- fixtures + const ( + address = "plataforma@everywhere.test" + password = "a long enough password" + ) + hash, err := argon2id.CreateHash(password, argon2id.DefaultParams) + if err != nil { + t.Fatalf("hash: %v", err) + } + domainID, mailboxID := uuid.New(), uuid.New() + pool.Exec(ctx, `INSERT INTO domains (id, name, punycode_name) VALUES ($1, 'everywhere.test', 'everywhere.test')`, domainID) + pool.Exec(ctx, `INSERT INTO mailboxes (id, domain_id, local_part, email_address, password_hash) VALUES ($1, $2, 'plataforma', $3, $4)`, + mailboxID, domainID, address, hash) + // The standard folders come from the one definition in the database + if _, err := pool.Exec(ctx, `SELECT provision_default_folders($1)`, mailboxID); err != nil { + t.Fatalf("provision folders: %v", err) + } + + blobs := diskstorage.NewLocalDiskBlobStorage(pool, t.TempDir()) + blobReader := diskstorage.NewLocalDiskBlobReader(pool) + authRepo := postgres.NewAuthRepository(pool) + messages := postgres.NewInboundMessageRepository(pool) + + // Three messages arrive from outside, as the SMTP side would file them + inbound := appusecase.NewProcessInboundEmailUseCase(postgres.NewMailboxRepository(pool), blobs, messages) + target, err := postgres.NewMailboxRepository(pool).FindActiveByAddress(ctx, address) + if err != nil || target == nil { + t.Fatalf("resolve mailbox: %v", err) + } + for i := 1; i <= 3; i++ { + raw := fmt.Sprintf("From: alice@remote.test\r\nTo: %s\r\nSubject: inbound %d\r\nMessage-ID: \r\n\r\nbody %d\r\n", address, i, i, i) + if err := inbound.Handle(ctx, appusecase.ProcessInboundEmailInput{ + Sender: "alice@remote.test", Recipients: []port.MailboxRecord{*target}, RecipientAddresses: []string{address}, Body: strings.NewReader(raw), + }); err != nil { + t.Fatalf("deliver %d: %v", i, err) + } + } + + // -------------------------------------------- the three ways in, wired as main.go + submissionUC := appusecase.NewProcessOutboundEmailUseCase(authRepo, postgres.NewOutboundRepository(pool), blobs, dkim.NewSigner(nil)) + submissionUC.SetSentCopies(messages) + + smtpServer := gosmtp.NewServer(smtppresentation.NewSubmissionBackend(submissionUC)) + smtpServer.Domain = "mail.everywhere.test" + smtpServer.TLSConfig = &tls.Config{Certificates: []tls.Certificate{selfSignedTLSCert(t)}} + smtpLn, _ := net.Listen("tcp", "127.0.0.1:0") + defer smtpLn.Close() + go smtpServer.Serve(smtpLn) + + imapUC := appusecase.NewImapSessionUseCase( + authRepo, postgres.NewImapFolderRepository(pool), postgres.NewMessageQueryRepository(pool), + postgres.NewFlagRepository(pool), blobReader, postgres.NewExpungeRepository(pool), postgres.NewCopyRepository(pool)) + imapServer := imapserver.New(&imapserver.Options{ + NewSession: func(c *imapserver.Conn) (imapserver.Session, *imapserver.GreetingData, error) { + return imappresentation.NewSession(c, imapUC) + }, + Caps: imap.CapSet{imap.CapIMAP4rev1: {}, imap.CapMove: {}, imap.CapUIDPlus: {}}, + InsecureAuth: true, + }) + imapLn, _ := net.Listen("tcp", "127.0.0.1:0") + go imapServer.Serve(imapLn) + defer imapServer.Close() + + webmailUC := appusecase.NewWebmailUseCase(postgres.NewWebmailRepository(pool), blobReader, submissionUC, authRepo, "mail.everywhere.test"). + WithLocalFiling(blobs, messages) + router := httppresentation.NewRouter(nil, func() error { return nil }) + router.SetMailAPI(webmailUC, webmailSecret) + web := httptest.NewServer(router) + defer web.Close() + token := mintSession(t, address, "user", time.Hour) + + // --------------------------------------------------------- small drivers + webCall := func(method, path string, body any) []byte { + t.Helper() + var reader *bytes.Reader + if body != nil { + encoded, _ := json.Marshal(body) + reader = bytes.NewReader(encoded) + } else { + reader = bytes.NewReader(nil) + } + req, _ := http.NewRequest(method, web.URL+path, reader) + req.Header.Set("Cookie", "lm_user_session="+token) + req.Header.Set("Content-Type", "application/json") + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("webmail %s %s: %v", method, path, err) + } + defer resp.Body.Close() + buf := new(bytes.Buffer) + buf.ReadFrom(resp.Body) + if resp.StatusCode >= 300 { + t.Fatalf("webmail %s %s = %d: %s", method, path, resp.StatusCode, buf.String()) + } + return buf.Bytes() + } + + type webFolder struct { + Name string `json:"name"` + UnreadCount int `json:"unread_count"` + TotalCount int `json:"total_count"` + } + webFolders := func() map[string]webFolder { + t.Helper() + var payload struct { + Folders []webFolder `json:"folders"` + } + raw := webCall(http.MethodGet, "/api/v1/mail/folders", nil) + if err := json.Unmarshal(raw, &payload); err != nil || payload.Folders == nil { + var plain []webFolder + if err2 := json.Unmarshal(raw, &plain); err2 != nil { + t.Fatalf("decode folders: %v / %v: %s", err, err2, raw) + } + payload.Folders = plain + } + out := map[string]webFolder{} + for _, f := range payload.Folders { + out[f.Name] = f + } + return out + } + type webMessage struct { + UID uint32 `json:"uid"` + Subject string `json:"subject"` + Seen bool `json:"seen"` + } + webMessages := func(folder string) []webMessage { + t.Helper() + var payload struct { + Messages []webMessage `json:"messages"` + } + raw := webCall(http.MethodGet, "/api/v1/mail/messages?folder="+folder, nil) + if err := json.Unmarshal(raw, &payload); err != nil || payload.Messages == nil { + var plain []webMessage + if err2 := json.Unmarshal(raw, &plain); err2 != nil { + t.Fatalf("decode messages: %v / %v: %s", err, err2, raw) + } + payload.Messages = plain + } + return payload.Messages + } + + imapClient, err := imapclient.DialInsecure(imapLn.Addr().String(), nil) + if err != nil { + t.Fatalf("dial imap: %v", err) + } + defer imapClient.Close() + if err := imapClient.Login(address, password).Wait(); err != nil { + t.Fatalf("imap login: %v", err) + } + imapExists := func(folder string) int { + t.Helper() + data, err := imapClient.Select(folder, nil).Wait() + if err != nil { + t.Fatalf("imap select %s: %v", folder, err) + } + return int(data.NumMessages) + } + imapFlags := func(folder string) map[uint32][]imap.Flag { + t.Helper() + imapExists(folder) + all := imap.UIDSet{} + all.AddRange(1, 0) // 1:* + cmd := imapClient.Fetch(all, &imap.FetchOptions{UID: true, Flags: true}) + out := map[uint32][]imap.Flag{} + msgs, err := cmd.Collect() + if err != nil { + t.Fatalf("imap fetch flags in %s: %v", folder, err) + } + for _, m := range msgs { + out[uint32(m.UID)] = m.Flags + } + return out + } + + // agree compares what the webmail and IMAP each say about every folder, and + // both with the rows: the one thing that must never differ + agree := func(after string) { + t.Helper() + assertEverywhere := webFolders() + for _, name := range []string{"INBOX", "Sent", "Trash", "Drafts"} { + if got, want := imapExists(name), assertEverywhere[name].TotalCount; got != want { + t.Errorf("after %s: %s holds %d messages over IMAP and %d in the webmail", after, name, got, want) + } + } + var badFolders, badBytes int + pool.QueryRow(ctx, ` + SELECT COUNT(*) FROM ( + SELECT f.id FROM folders f + LEFT JOIN email_messages e ON e.folder_id = f.id AND e.expunged_at IS NULL + WHERE f.mailbox_id = $1 + GROUP BY f.id, f.total_count, f.unread_count + HAVING f.total_count <> COUNT(e.id) + OR f.unread_count <> COUNT(e.id) FILTER (WHERE NOT EXISTS ( + SELECT 1 FROM message_flags mf WHERE mf.message_id = e.id AND mf.received_at = e.received_at AND mf.flag = '\Seen')) + ) x`, mailboxID).Scan(&badFolders) + pool.QueryRow(ctx, ` + SELECT COUNT(*) FROM mailboxes m + WHERE m.id = $1 AND m.used_bytes <> COALESCE((SELECT SUM(size_bytes) FROM email_messages WHERE mailbox_id = m.id AND expunged_at IS NULL), 0)`, + mailboxID).Scan(&badBytes) + if badFolders != 0 { + t.Errorf("after %s: %d folder(s) have cached counters that differ from their messages", after, badFolders) + } + if badBytes != 0 { + t.Errorf("after %s: the quota differs from the bytes actually stored", after) + } + } + + agree("delivery") + if f := webFolders(); f["INBOX"].UnreadCount != 3 || f["INBOX"].TotalCount != 3 { + t.Fatalf("INBOX should start with 3 unread of 3, the webmail says %+v", f["INBOX"]) + } + + // ----- 1. SMTP sends a message; Sent must show it in the webmail and over IMAP + smtpClient, err := gosmtp.DialStartTLS(smtpLn.Addr().String(), &tls.Config{InsecureSkipVerify: true}) + if err != nil { + t.Fatalf("dial smtp: %v", err) + } + if err := smtpClient.Auth(plainAuth(address, password)); err != nil { + t.Fatalf("smtp auth: %v", err) + } + smtpClient.Mail(address, nil) + smtpClient.Rcpt("bob@remote.test", nil) + w, _ := smtpClient.Data() + w.Write([]byte("From: " + address + "\r\nTo: bob@remote.test\r\nSubject: sent over smtp\r\nMessage-ID: \r\n\r\nhello\r\n")) + if err := w.Close(); err != nil { + t.Fatalf("smtp data: %v", err) + } + smtpClient.Quit() + + agree("sending over SMTP") + if sent := webMessages("sent"); len(sent) != 1 || sent[0].Subject != "sent over smtp" { + t.Fatalf("the webmail's Sent should hold the SMTP message, got %+v", sent) + } + if n := imapExists("Sent"); n != 1 { + t.Fatalf("IMAP's Sent holds %d messages after an SMTP send, want 1", n) + } + if f := webFolders(); f["Sent"].UnreadCount != 0 { + t.Errorf("the sender's own copy shows as unread: %+v", f["Sent"]) + } + + // ----- 2. the webmail sends one too: it must land in the same place, once + webCall(http.MethodPost, "/api/v1/mail/send", map[string]any{"to": []string{"carol@remote.test"}, "subject": "sent from webmail", "text": "hi"}) + agree("sending from the webmail") + if n := imapExists("Sent"); n != 2 { + t.Fatalf("Sent holds %d messages after one SMTP send and one webmail send, want exactly 2", n) + } + + // ----- 3. read in the webmail; IMAP must see \Seen and the badge must fall + inbox := webMessages("inbox") + if len(inbox) != 3 { + t.Fatalf("webmail lists %d INBOX messages, want 3", len(inbox)) + } + uidA, uidB, uidC := inbox[2].UID, inbox[1].UID, inbox[0].UID // oldest first by UID + webCall(http.MethodPost, "/api/v1/mail/seen", map[string]any{"folder": "inbox", "uid": uidA, "seen": true}) + agree("reading a message in the webmail") + if flags := imapFlags("INBOX")[uidA]; !hasFlag(flags, imap.FlagSeen) { + t.Errorf("IMAP does not see the message the webmail marked as read: %v", flags) + } + if f := webFolders(); f["INBOX"].UnreadCount != 2 { + t.Errorf("INBOX unread = %d after one read, want 2", f["INBOX"].UnreadCount) + } + + // ----- 4. read over IMAP; the webmail's badge must follow (it never did) + imapExists("INBOX") + if err := imapClient.Store(imap.UIDSetNum(imap.UID(uidB)), &imap.StoreFlags{Op: imap.StoreFlagsAdd, Silent: true, Flags: []imap.Flag{imap.FlagSeen}}, nil).Close(); err != nil { + t.Fatalf("imap store: %v", err) + } + agree("reading a message over IMAP") + if f := webFolders(); f["INBOX"].UnreadCount != 1 { + t.Errorf("INBOX unread = %d after a read over IMAP, want 1: the badge ignored it", f["INBOX"].UnreadCount) + } + for _, m := range webMessages("inbox") { + if m.UID == uidB && !m.Seen { + t.Error("the webmail lists the message IMAP marked as read as unread") + } + } + + // ----- 5. delete over IMAP (\Deleted + EXPUNGE); the webmail list shrinks, the quota is returned + imapExists("INBOX") + if err := imapClient.Store(imap.UIDSetNum(imap.UID(uidC)), &imap.StoreFlags{Op: imap.StoreFlagsAdd, Silent: true, Flags: []imap.Flag{imap.FlagDeleted}}, nil).Close(); err != nil { + t.Fatalf("imap store deleted: %v", err) + } + if err := imapClient.Expunge().Close(); err != nil { + t.Fatalf("imap expunge: %v", err) + } + agree("deleting over IMAP") + if n := len(webMessages("inbox")); n != 2 { + t.Errorf("the webmail lists %d INBOX messages after an IMAP delete, want 2", n) + } + + // ----- 6. delete in the webmail (to Trash); IMAP sees it in Trash and not in INBOX + webCall(http.MethodPost, "/api/v1/mail/delete", map[string]any{"folder": "inbox", "uid": uidA}) + agree("deleting in the webmail") + if n := imapExists("Trash"); n != 1 { + t.Errorf("IMAP's Trash holds %d messages, want 1", n) + } + if n := imapExists("INBOX"); n != 1 { + t.Errorf("IMAP's INBOX holds %d messages, want 1", n) + } + + // ----- 7. empty the Trash in the webmail: the bytes go back to the quota + trash := webMessages("trash") + if len(trash) != 1 { + t.Fatalf("the webmail lists %d Trash messages, want 1", len(trash)) + } + webCall(http.MethodPost, "/api/v1/mail/delete", map[string]any{"folder": "trash", "uid": trash[0].UID}) + agree("emptying the Trash") + if n := imapExists("Trash"); n != 0 { + t.Errorf("IMAP's Trash holds %d messages after emptying it, want 0", n) + } +} + +func hasFlag(flags []imap.Flag, want imap.Flag) bool { + for _, f := range flags { + if f == want { + return true + } + } + return false +} diff --git a/services/protocols/internal/infrastructure/postgres/mail_state_contract_test.go b/services/protocols/internal/infrastructure/postgres/mail_state_contract_test.go index fb80f66..54f041c 100644 --- a/services/protocols/internal/infrastructure/postgres/mail_state_contract_test.go +++ b/services/protocols/internal/infrastructure/postgres/mail_state_contract_test.go @@ -403,3 +403,132 @@ func TestContract_ReplacingTheFlagSetKeepsTheUnreadCounterTrue(t *testing.T) { } var _ = uuid.Nil + +// IMAP clients, the webmail and the delivery path all touch a mailbox at the same +// moment. The counters are shared rows, so they have to stay true under that. +func TestContract_ConcurrentChangesFromBothPathsKeepTheCountersTrue(t *testing.T) { + pool := testPool(t) + mailboxID := seedMailbox(t, pool) + + const n = 24 + uids := make([]uint32, 0, n) + for i := 0; i < n; i++ { + uids = append(uids, deliver(t, pool, mailboxID, "INBOX", fmt.Sprintf("m%d", i), false)) + } + + imap, web := imapPath(), webmailPath() + done := make(chan struct{}, n) + for i, uid := range uids { + go func(i int, uid uint32) { + defer func() { done <- struct{}{} }() + switch i % 3 { + case 0: + imap.setSeen(t, pool, mailboxID, "INBOX", uid, true) + case 1: + web.setSeen(t, pool, mailboxID, "INBOX", uid, true) + default: + if i%2 == 0 { + imap.expunge(t, pool, mailboxID, "INBOX", uid) + } else { + web.expunge(t, pool, mailboxID, "INBOX", uid) + } + } + }(i, uid) + } + for range uids { + <-done + } + + assertConsistent(t, pool, mailboxID, "24 concurrent changes from IMAP and the webmail") +} + +// Two clients deleting the same message at once is the realistic race (the +// webmail and a phone both open on the same mail). It is removed once, and every +// number moves once. +func TestContract_DeletingTheSameMessageConcurrentlyCountsOnce(t *testing.T) { + pool := testPool(t) + mailboxID := seedMailbox(t, pool) + target := deliver(t, pool, mailboxID, "INBOX", "contested", false) + deliver(t, pool, mailboxID, "INBOX", "bystander", false) + inbox := folderID(t, pool, mailboxID, "INBOX") + + ctx := context.Background() + store := NewMailStore(pool) + removed := make(chan int, 12) + for i := 0; i < 12; i++ { + go func() { + n, err := store.Expunge(ctx, inbox, []uint32{target}) + if err != nil { + t.Errorf("expunge: %v", err) + } + removed <- n + }() + } + total := 0 + for i := 0; i < 12; i++ { + total += <-removed + } + + if total != 1 { + t.Errorf("12 concurrent expunges of one message removed it %d times, want exactly once", total) + } + assertConsistent(t, pool, mailboxID, "12 clients deleting the same message") + folders, used, _ := mailState(t, pool, mailboxID) + if folders["INBOX"].RealTotal != 1 || used != 10 { + t.Errorf("INBOX holds %d and the mailbox %d bytes, want 1 message and 10 bytes", folders["INBOX"].RealTotal, used) + } +} + +// Moves in opposite directions, flag changes and deletes, all at once on the same +// two folders. Locking the folders and the message in different orders in +// different operations deadlocks here; a fixed order does not. +func TestContract_OppositeMovesAndFlagChangesNeverDeadlock(t *testing.T) { + pool := testPool(t) + mailboxID := seedMailbox(t, pool) + + const each = 20 + var inboxUIDs, trashUIDs []uint32 + for i := 0; i < each; i++ { + inboxUIDs = append(inboxUIDs, deliver(t, pool, mailboxID, "INBOX", fmt.Sprintf("i%d", i), false)) + trashUIDs = append(trashUIDs, deliver(t, pool, mailboxID, "Trash", fmt.Sprintf("t%d", i), false)) + } + + inbox, trash := folderID(t, pool, mailboxID, "INBOX"), folderID(t, pool, mailboxID, "Trash") + store := NewMailStore(pool) + ctx := context.Background() + + errs := make(chan error, each*4) + var pending int + run := func(f func() error) { + pending++ + go func() { errs <- f() }() + } + for i := 0; i < each; i++ { + i := i + run(func() error { _, err := store.Move(ctx, inbox, inboxUIDs[i], trash); return nil2(err) }) + run(func() error { _, err := store.Move(ctx, trash, trashUIDs[i], inbox); return nil2(err) }) + // a flag change on a message that is being moved elsewhere, and a delete + run(func() error { + _, err := store.SetFlags(ctx, inbox, inboxUIDs[(i+1)%each], port.FlagOpAdd, []string{`\Flagged`}, 0) + return nil2(err) + }) + run(func() error { _, err := store.Expunge(ctx, trash, []uint32{trashUIDs[(i+2)%each]}); return nil2(err) }) + } + for ; pending > 0; pending-- { + if err := <-errs; err != nil { + t.Errorf("a concurrent operation failed (a deadlock, if it says 40P01): %v", err) + } + } + + assertConsistent(t, pool, mailboxID, "opposite moves, flag changes and deletes all at once") +} + +// nil2 lets the contract tolerate "the message was moved or deleted first", a +// legitimate outcome of a race, while still surfacing every other error (above +// all a deadlock, SQLSTATE 40P01). +func nil2(err error) error { + if err == ErrMessageNotFound { + return nil + } + return err +} diff --git a/services/protocols/internal/infrastructure/postgres/mail_store.go b/services/protocols/internal/infrastructure/postgres/mail_store.go index f529dc5..1a0fb30 100644 --- a/services/protocols/internal/infrastructure/postgres/mail_store.go +++ b/services/protocols/internal/infrastructure/postgres/mail_store.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "sort" "github.com/google/uuid" "github.com/jackc/pgx/v5" @@ -44,6 +45,27 @@ func NewMailStore(pool *pgxpool.Pool) *MailStore { return &MailStore{pool: pool} } +// lockFolders takes the row lock of each folder, in a fixed order. +// +// Every change updates counters on a folder row, and a move updates two. If one +// operation locked the destination folder and then the source, and another did +// the reverse, two moves in opposite directions deadlocked, and so did a move +// racing a flag change (it held the message and wanted the folder, the move held +// the folder and wanted the message). So the order is the same everywhere: +// folders first, lowest id first, then the message. +func lockFolders(ctx context.Context, tx pgx.Tx, ids ...string) error { + sort.Strings(ids) + for i, id := range ids { + if i > 0 && id == ids[i-1] { + continue + } + if _, err := tx.Exec(ctx, `SELECT 1 FROM folders WHERE id = $1 FOR UPDATE`, id); err != nil { + return fmt.Errorf("lock folder %s: %w", id, err) + } + } + return nil +} + // messageRow is the slice of a message every operation needs. type messageRow struct { ID uuid.UUID @@ -110,6 +132,9 @@ func (s *MailStore) SetFlags(ctx context.Context, folderID string, uid uint32, o } defer tx.Rollback(ctx) + if err := lockFolders(ctx, tx, folderID); err != nil { + return false, err + } msg, err := loadMessage(ctx, tx, folderID, uid) if err != nil { return false, err @@ -239,6 +264,10 @@ func (s *MailStore) Expunge(ctx context.Context, folderID string, uids []uint32) } defer tx.Rollback(ctx) + if err := lockFolders(ctx, tx, folderID); err != nil { + return 0, err + } + removed := 0 for _, uid := range uids { msg, err := loadMessage(ctx, tx, folderID, uid) @@ -288,14 +317,17 @@ func (s *MailStore) Move(ctx context.Context, sourceFolderID string, uid uint32, } defer tx.Rollback(ctx) + if err := lockFolders(ctx, tx, sourceFolderID, destFolderID); err != nil { + return 0, err + } msg, err := loadMessage(ctx, tx, sourceFolderID, uid) if err != nil { return 0, err } var destUID int64 - if err := tx.QueryRow(ctx, `SELECT uid_next FROM folders WHERE id = $1 FOR UPDATE`, destFolderID).Scan(&destUID); err != nil { - return 0, fmt.Errorf("lock destination folder: %w", err) + if err := tx.QueryRow(ctx, `SELECT uid_next FROM folders WHERE id = $1`, destFolderID).Scan(&destUID); err != nil { + return 0, fmt.Errorf("read destination uid: %w", err) } unread := 0 @@ -353,6 +385,9 @@ func (s *MailStore) copyOne(ctx context.Context, sourceFolderID string, sourceUI } defer tx.Rollback(ctx) + if err := lockFolders(ctx, tx, destFolderID); err != nil { + return 0, err + } src, err := loadMessage(ctx, tx, sourceFolderID, sourceUID) if err != nil { return 0, err @@ -360,8 +395,8 @@ func (s *MailStore) copyOne(ctx context.Context, sourceFolderID string, sourceUI var destUID int64 var destMailboxID string - if err := tx.QueryRow(ctx, `SELECT uid_next, mailbox_id::text FROM folders WHERE id = $1 FOR UPDATE`, destFolderID).Scan(&destUID, &destMailboxID); err != nil { - return 0, fmt.Errorf("lock destination folder: %w", err) + if err := tx.QueryRow(ctx, `SELECT uid_next, mailbox_id::text FROM folders WHERE id = $1`, destFolderID).Scan(&destUID, &destMailboxID); err != nil { + return 0, fmt.Errorf("read destination folder: %w", err) } unread := 0 From f2951c0ff4cf1825c9986cb24bf51bc07ecb678f Mon Sep 17 00:00:00 2001 From: Lucas Rayan Date: Fri, 16 Oct 2026 13:36:21 -0300 Subject: [PATCH 6/6] docs: describe the mailbox core, its invariants and lock order Co-Authored-By: Claude Sonnet 5.5 --- docs/MAILBOX-CORE.md | 57 ++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 57 insertions(+) create mode 100644 docs/MAILBOX-CORE.md diff --git a/docs/MAILBOX-CORE.md b/docs/MAILBOX-CORE.md new file mode 100644 index 0000000..85acbe9 --- /dev/null +++ b/docs/MAILBOX-CORE.md @@ -0,0 +1,57 @@ +# Mailbox core + +The webmail, IMAP, POP3 and SMTP are four ways in to one mailbox. They share +the rules and the data code; none of them has its own copy. + +## Where each rule lives + +| Rule | One place | +|------|-----------| +| Flags, expunge, move, copy, folder delete, counters, quota | `MailStore` (`services/protocols/internal/infrastructure/postgres/mail_store.go`) | +| Folder and message reads | `mail_store_read.go` (same store) | +| Credentials (password, app password, scopes) | `usecase/credentials.go` | +| Sending, and the copy filed in Sent | `ProcessOutboundEmail.Submit` (`FileSentCopy`) | +| Default folders of a mailbox | SQL function `provision_default_folders` (migration 0014) | + +The IMAP repositories, the webmail repository and the TS auth service are thin +adapters: they translate their protocol and call the code above. + +## Invariants + +After any operation, through any port: + +- `folders.total_count` and `unread_count` equal the live messages and the ones + without `\Seen`. +- `folders.highest_modseq` rises only when a message really changed. +- `mailboxes.used_bytes` equals the size of the live messages. +- Deleting twice, or from two ports at once, counts once. + +`mail_state_contract_test.go` checks all of this after every operation through +the IMAP path and the webmail path, and under concurrency. +`e2e/one_mailbox_everywhere_e2e_test.go` does the same over real SMTP, IMAP and +HTTP on one mailbox. + +## Lock order + +Every operation locks **folders first, lowest id first, then the message** +(`lockFolders`). Two moves in opposite directions, or a move racing a flag +change, used to deadlock (SQLSTATE 40P01) because they locked in different +orders. `TestContract_OppositeMovesAndFlagChangesNeverDeadlock` covers it. A new +operation must follow the same order. + +## Adding an operation + +1. Put it in `MailStore`, in one transaction, with `lockFolders`. +2. Make the adapters call it. +3. Add it to the contract test and run it through both paths. + +## Migration 0014 + +Creates `provision_default_folders`, creates the missing default folders of +existing mailboxes, and recomputes folder counters and `used_bytes` from the +messages. It is idempotent and runs with the other migrations at deploy. + +## Not done + +IMAP `APPEND` is still a stub. When it is implemented, drop a client-appended +Sent copy that matches the server's one by `Message-ID`.