lucasrguerra/LambdaMail
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

#6: implement IMAP IDLE (RFC 2177) and CONDSTORE (RFC 7162)

Browse Source
This commit is contained in:
Lucas Rayan committed 2026-07-24 11:43:37 -03:00
1 parent 0ec0698079
commit 6b1e4dc00a
20 files changed
+539 -95
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

services/protocols/internal/infrastructure/postgres/copy_repository.go
+6 -5
View File
@@ -72,8 +72,9 @@ func (r *CopyRepository) copyOne(ctx context.Context, sourceFolderID string, sou
}
destUID := uint32(destUIDNext)
if _, err := tx.Exec(ctx, `UPDATE folders SET uid_next = uid_next + 1, total_count = total_count + 1 WHERE id = $1`, destFolderID); err != nil {
return 0, fmt.Errorf("advance destination uid_next: %w", err)
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
@@ -97,14 +98,14 @@ func (r *CopyRepository) copyOne(ctx context.Context, sourceFolderID string, sou
var destReceivedAt any
err = tx.QueryRow(ctx, `
INSERT INTO email_messages (
mailbox_id, folder_id, uid, blob_id,
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)
) 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, blobID, senderAddress, recipientAddresses, sizeBytes, spfResult, dkimResult, dmarcResult,
`, 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 {
services/protocols/internal/infrastructure/postgres/expunge_repository.go
+1 -1
View File
@@ -71,7 +71,7 @@ func (r *ExpungeRepository) Expunge(ctx context.Context, folderID string, uids [
unreadDelta = 1
}
if _, err := tx.Exec(ctx, `
UPDATE folders SET total_count = total_count - 1, unread_count = unread_count - $1
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)
services/protocols/internal/infrastructure/postgres/flag_repository.go
+44 -20
View File
@@ -18,55 +18,79 @@ func NewFlagRepository(pool *pgxpool.Pool) *FlagRepository {
return &FlagRepository{pool: pool}
}
func (r *FlagRepository) SetFlags(ctx context.Context, folderID string, uid uint32, op port.FlagOp, flags []string) error {
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 fmt.Errorf("begin transaction: %w", err)
return false, fmt.Errorf("begin transaction: %w", err)
}
defer tx.Rollback(ctx)
var messageID string
var receivedAt any
var currentModSeq uint64
if err := tx.QueryRow(ctx, `
SELECT id, received_at FROM email_messages WHERE folder_id = $1 AND uid = $2
`, folderID, uid).Scan(&messageID, &receivedAt); err != nil {
return fmt.Errorf("find message folder=%s uid=%d: %w", folderID, uid, err)
SELECT modseq FROM email_messages WHERE folder_id = $1 AND uid = $2 AND expunged_at IS NULL
`, folderID, uid).Scan(&currentModSeq); 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) VALUES ($1, $2, $3)
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
`, messageID, receivedAt, flag); err != nil {
return fmt.Errorf("add flag %q: %w", flag, err)
`, 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 = $1 AND received_at = $2 AND flag = $3
`, messageID, receivedAt, flag); err != nil {
return fmt.Errorf("delete flag %q: %w", flag, err)
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 = $1 AND received_at = $2`, messageID, receivedAt); err != nil {
return fmt.Errorf("clear flags before set: %w", err)
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) VALUES ($1, $2, $3)
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
`, messageID, receivedAt, flag); err != nil {
return fmt.Errorf("set flag %q: %w", flag, err)
`, folderID, uid, flag); err != nil {
return false, fmt.Errorf("set flag %q: %w", flag, err)
}
}
}
if err := tx.Commit(ctx); err != nil {
return fmt.Errorf("commit: %w", err)
return false, fmt.Errorf("commit: %w", err)
}
return nil
return true, nil
}
services/protocols/internal/infrastructure/postgres/flag_repository_test.go
+19 -8
View File
@@ -70,8 +70,8 @@ func TestFlagRepository_SetFlags_AddThenDelThenSet(t *testing.T) {
repo := NewFlagRepository(pool)
if err := repo.SetFlags(ctx, folderID.String(), 7, port.FlagOpAdd, []string{"\\Seen", "\\Flagged"}); err != nil {
t.Fatalf("add: %v", err)
if updated, err := repo.SetFlags(ctx, folderID.String(), 7, port.FlagOpAdd, []string{"\\Seen", "\\Flagged"}, 0); err != nil || !updated {
t.Fatalf("add: updated=%v, err=%v", updated, err)
}
got := readFlags()
sort.Strings(got)
@@ -79,16 +79,16 @@ func TestFlagRepository_SetFlags_AddThenDelThenSet(t *testing.T) {
t.Fatalf("after add: flags = %v, want [\\Flagged \\Seen]", got)
}
if err := repo.SetFlags(ctx, folderID.String(), 7, port.FlagOpDel, []string{"\\Flagged"}); err != nil {
t.Fatalf("del: %v", err)
if updated, err := repo.SetFlags(ctx, folderID.String(), 7, port.FlagOpDel, []string{"\\Flagged"}, 0); err != nil || !updated {
t.Fatalf("del: updated=%v, err=%v", updated, err)
}
got = readFlags()
if len(got) != 1 || got[0] != "\\Seen" {
t.Fatalf("after del: flags = %v, want [\\Seen]", got)
}
if err := repo.SetFlags(ctx, folderID.String(), 7, port.FlagOpSet, []string{"\\Answered"}); err != nil {
t.Fatalf("set: %v", err)
if updated, err := repo.SetFlags(ctx, folderID.String(), 7, port.FlagOpSet, []string{"\\Answered"}, 0); err != nil || !updated {
t.Fatalf("set: updated=%v, err=%v", updated, err)
}
got = readFlags()
if len(got) != 1 || got[0] != "\\Answered" {
@@ -99,11 +99,22 @@ func TestFlagRepository_SetFlags_AddThenDelThenSet(t *testing.T) {
// single command - must not fail the whole STORE with a primary key
// violation on the second INSERT for the same (message_id, received_at,
// flag).
if err := repo.SetFlags(ctx, folderID.String(), 7, port.FlagOpSet, []string{"\\Seen", "\\Seen"}); err != nil {
t.Fatalf("set with duplicate flags in input: %v", err)
if updated, err := repo.SetFlags(ctx, folderID.String(), 7, port.FlagOpSet, []string{"\\Seen", "\\Seen"}, 0); err != nil || !updated {
t.Fatalf("set with duplicate flags in input: updated=%v, err=%v", updated, err)
}
got = readFlags()
if len(got) != 1 || got[0] != "\\Seen" {
t.Fatalf("after set with duplicate input flags: flags = %v, want [\\Seen]", got)
}
// Test UNCHANGEDSINCE conditional store:
// Message's modseq was updated during the previous SetFlags call.
// Providing a lower unchangedSince value must fail (return updated=false).
if updated, err := repo.SetFlags(ctx, folderID.String(), 7, port.FlagOpSet, []string{"\\Flagged"}, 1); err != nil || updated {
t.Fatalf("conditional set with lower modseq: expected updated=false, got updated=%v, err=%v", updated, err)
}
// Providing a matching or higher unchangedSince value must succeed (return updated=true).
if updated, err := repo.SetFlags(ctx, folderID.String(), 7, port.FlagOpSet, []string{"\\Flagged"}, 100); err != nil || !updated {
t.Fatalf("conditional set with higher modseq: expected updated=true, got updated=%v, err=%v", updated, err)
}
}
services/protocols/internal/infrastructure/postgres/imap_folder_repository.go
+4 -4
View File
@@ -20,13 +20,13 @@ func NewImapFolderRepository(pool *pgxpool.Pool) *ImapFolderRepository {
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
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); err != nil {
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
}
@@ -37,7 +37,7 @@ func (r *ImapFolderRepository) FindByName(ctx context.Context, mailboxID, name s
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
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
@@ -50,7 +50,7 @@ func (r *ImapFolderRepository) ListFolders(ctx context.Context, mailboxID string
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); err != nil {
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)
services/protocols/internal/infrastructure/postgres/inbound_message_repository.go
+9 -7
View File
@@ -39,21 +39,23 @@ func (r *InboundMessageRepository) Persist(ctx context.Context, input port.Persi
return 0, fmt.Errorf("find inbox folder for mailbox %s: %w", input.MailboxID, err)
}
if _, err := tx.Exec(ctx, `
UPDATE folders SET uid_next = uid_next + 1, unread_count = unread_count + 1, total_count = total_count + 1
var modseq int64
if err := tx.QueryRow(ctx, `
UPDATE folders SET uid_next = uid_next + 1, highest_modseq = highest_modseq + 1, unread_count = unread_count + 1, total_count = total_count + 1
WHERE id = $1
`, folderID); err != nil {
return 0, fmt.Errorf("advance uid_next and folder counters: %w", err)
RETURNING highest_modseq
`, folderID).Scan(&modseq); err != nil {
return 0, fmt.Errorf("advance uid_next, highest_modseq and folder counters: %w", err)
}
// email_messages.id defaults to gen_random_uuid() per the migration - no explicit value needed.
_, err = tx.Exec(ctx, `
INSERT INTO email_messages (
mailbox_id, folder_id, uid, blob_id,
mailbox_id, folder_id, uid, modseq, blob_id,
sender_address, recipient_addresses, size_bytes,
spf_result, dkim_result, dmarc_result
) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
`, input.MailboxID, folderID, uid, input.Blob.ID,
) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)
`, input.MailboxID, folderID, uid, modseq, input.Blob.ID,
input.SenderAddress, []string{input.RecipientAddress}, input.Blob.SizeBytes,
input.SPFResult, input.DKIMResult, input.DMARCResult)
if err != nil {
services/protocols/internal/infrastructure/postgres/message_query_repository.go
+3 -3
View File
@@ -19,12 +19,12 @@ func NewMessageQueryRepository(pool *pgxpool.Pool) *MessageQueryRepository {
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,
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
GROUP BY m.uid, m.blob_id, m.size_bytes, m.received_at, m.modseq
ORDER BY m.uid ASC
`, folderID)
if err != nil {
@@ -35,7 +35,7 @@ func (r *MessageQueryRepository) ListMessages(ctx context.Context, folderID stri
var out []port.MessageRecord
for rows.Next() {
var rec port.MessageRecord
if err := rows.Scan(&rec.UID, &rec.BlobID, &rec.SizeBytes, &rec.ReceivedAt, &rec.Flags); err != nil {
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)