From caf52c02257099d8059a971157b25d9647ad6ce9 Mon Sep 17 00:00:00 2001 From: Lucas Rayan Date: Fri, 16 Oct 2026 13:15:32 -0300 Subject: [PATCH] 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) }