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