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.
This commit is contained in:
Lucas Rayan committed
1 parent
e63b68acb2
commit
caf52c0225
5 files changed
+550
-593
No files matched your search
@@ -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)
|
||||
}
|
||||
Reference in new issue
Block a user