#5: F2 sub-project 2 - IMAP SEARCH, EXPUNGE, COPY, MOVE

Adds the second F2 sub-project on top of the merged IMAP core: full SEARCH criteria matching (flags, dates, size, headers, body/text, NOT/OR), EXPUNGE (soft-delete via expunged_at), COPY, and MOVE (RFC 6851, advertised via CapMove/CapUIDPlus). Verified end-to-end with a real go-imap/v2 client against embedded Postgres. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-07-24 09:40:47 -03:00
1 parent de082a190c
commit 0ec0698079
@@ -0,0 +1,133 @@
+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).
+type CopyRepository struct {
+	pool *pgxpool.Pool
+}
+
+func NewCopyRepository(pool *pgxpool.Pool) *CopyRepository {
+	return &CopyRepository{pool: 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)
+
+	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 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, 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)
+		RETURNING id, received_at
+	`, destMailboxID, destFolderID, destUID, 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
+}
@@ -0,0 +1,110 @@
+package postgres
+
+import (
+	"context"
+	"testing"
+
+	"github.com/google/uuid"
+	"lambdamail/protocols/internal/application/port"
+)
+
+func TestCopyRepository_CopyMessages_AllocatesNewUIDsAndPreservesFlags(t *testing.T) {
+	pool := testPool(t)
+	defer pool.Close()
+	ctx := context.Background()
+
+	domainID := uuid.New()
+	domainName := "copyrepo-test-" + domainID.String() + ".invalid"
+	_, err := pool.Exec(ctx, `INSERT INTO domains (id, name, punycode_name) VALUES ($1, $2, $3)`,
+		domainID, domainName, domainName)
+	if err != nil {
+		t.Fatalf("seed domain: %v", err)
+	}
+	defer pool.Exec(ctx, `DELETE FROM domains WHERE id = $1`, domainID)
+
+	mailboxID := uuid.New()
+	_, err = pool.Exec(ctx, `INSERT INTO mailboxes (id, domain_id, local_part, email_address, password_hash) VALUES ($1, $2, 'user', $3, '$argon2id$v=19$m=65536,t=3,p=4$c29tZXNhbHQ$RdescudvJCsgt3ub+b+dWRWJTmaaJObG')`,
+		mailboxID, domainID, "user@"+domainName)
+	if err != nil {
+		t.Fatalf("seed mailbox: %v", err)
+	}
+	sourceFolderID := uuid.New()
+	_, err = pool.Exec(ctx, `INSERT INTO folders (id, mailbox_id, name, special_use) VALUES ($1, $2, 'INBOX', 'inbox')`,
+		sourceFolderID, mailboxID)
+	if err != nil {
+		t.Fatalf("seed source folder: %v", err)
+	}
+	destFolderID := uuid.New()
+	_, err = pool.Exec(ctx, `INSERT INTO folders (id, mailbox_id, name) VALUES ($1, $2, 'Archive')`,
+		destFolderID, mailboxID)
+	if err != nil {
+		t.Fatalf("seed dest folder: %v", err)
+	}
+	blobID := uuid.New()
+	_, err = pool.Exec(ctx, `INSERT INTO message_blobs (id, content_sha256, storage_driver, storage_path, size_bytes, ref_count) VALUES ($1, $2, 'local', '/tmp/x', 10, 1)`,
+		blobID, uuid.New().String()+uuid.New().String()[:32])
+	if err != nil {
+		t.Fatalf("seed blob: %v", err)
+	}
+	const sourceSubject = "Original subject line"
+	var sourceMessageID string
+	if err := pool.QueryRow(ctx, `
+		INSERT INTO email_messages (mailbox_id, folder_id, uid, blob_id, sender_address, recipient_addresses, size_bytes, subject)
+		VALUES ($1, $2, 5, $3, 'sender@example.test', $4, 10, $5) RETURNING id
+	`, mailboxID, sourceFolderID, blobID, []string{"user@" + domainName}, sourceSubject).Scan(&sourceMessageID); err != nil {
+		t.Fatalf("seed source message: %v", err)
+	}
+	if _, err := pool.Exec(ctx, `
+		INSERT INTO message_flags (message_id, received_at, flag)
+		SELECT id, received_at, '\Seen' FROM email_messages WHERE id = $1
+	`, sourceMessageID); err != nil {
+		t.Fatalf("seed source flag: %v", err)
+	}
+
+	repo := NewCopyRepository(pool)
+	copied, err := repo.CopyMessages(ctx, sourceFolderID.String(), []uint32{5}, destFolderID.String())
+	if err != nil {
+		t.Fatalf("CopyMessages: %v", err)
+	}
+	if len(copied) != 1 || copied[0].SourceUID != 5 || copied[0].DestUID != 1 {
+		t.Fatalf("copied = %+v, want [{SourceUID:5 DestUID:1}]", copied)
+	}
+
+	var destCount int
+	if err := pool.QueryRow(ctx, `SELECT count(*) FROM email_messages WHERE folder_id = $1`, destFolderID).Scan(&destCount); err != nil {
+		t.Fatalf("count dest messages: %v", err)
+	}
+	if destCount != 1 {
+		t.Fatalf("dest folder has %d messages, want 1", destCount)
+	}
+
+	var destFlagCount int
+	if err := pool.QueryRow(ctx, `
+		SELECT count(*) 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 mf.flag = '\Seen'
+	`, destFolderID).Scan(&destFlagCount); err != nil {
+		t.Fatalf("count dest flags: %v", err)
+	}
+	if destFlagCount != 1 {
+		t.Errorf("dest message has %d \\Seen flags, want 1 (flags must be preserved on copy)", destFlagCount)
+	}
+
+	var destSubject *string
+	if err := pool.QueryRow(ctx, `SELECT subject FROM email_messages WHERE folder_id = $1`, destFolderID).Scan(&destSubject); err != nil {
+		t.Fatalf("query dest subject: %v", err)
+	}
+	if destSubject == nil || *destSubject != sourceSubject {
+		t.Errorf("dest subject = %v, want %q (subject must be preserved on copy)", destSubject, sourceSubject)
+	}
+
+	var refCount int
+	if err := pool.QueryRow(ctx, `SELECT ref_count FROM message_blobs WHERE id = $1`, blobID).Scan(&refCount); err != nil {
+		t.Fatalf("query ref_count: %v", err)
+	}
+	if refCount != 2 {
+		t.Errorf("ref_count = %d, want 2 (1 original + 1 copy referencing the same blob)", refCount)
+	}
+
+	_ = port.CopiedMessage{} // keep the port import used if the assertions above change
+}
@@ -0,0 +1,92 @@
+package postgres
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+// ExpungeRepository implements port.ExpungeRepository against Postgres.
+type ExpungeRepository struct {
+	pool *pgxpool.Pool
+}
+
+func NewExpungeRepository(pool *pgxpool.Pool) *ExpungeRepository {
+	return &ExpungeRepository{pool: 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
+			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
+}
@@ -0,0 +1,96 @@
+package postgres
+
+import (
+	"context"
+	"testing"
+
+	"github.com/google/uuid"
+)
+
+func TestExpungeRepository_Expunge_SetsExpungedAtAndDecrementsCounters(t *testing.T) {
+	pool := testPool(t)
+	defer pool.Close()
+	ctx := context.Background()
+
+	domainID := uuid.New()
+	domainName := "expunge-test-" + domainID.String() + ".invalid"
+	_, err := pool.Exec(ctx, `INSERT INTO domains (id, name, punycode_name) VALUES ($1, $2, $3)`,
+		domainID, domainName, domainName)
+	if err != nil {
+		t.Fatalf("seed domain: %v", err)
+	}
+	defer pool.Exec(ctx, `DELETE FROM domains WHERE id = $1`, domainID)
+
+	mailboxID := uuid.New()
+	_, err = pool.Exec(ctx, `INSERT INTO mailboxes (id, domain_id, local_part, email_address, password_hash, used_bytes) VALUES ($1, $2, 'user', $3, '$argon2id$v=19$m=65536,t=3,p=4$c29tZXNhbHQ$RdescudvJCsgt3ub+b+dWRWJTmaaJObG', 20)`,
+		mailboxID, domainID, "user@"+domainName)
+	if err != nil {
+		t.Fatalf("seed mailbox: %v", err)
+	}
+	folderID := uuid.New()
+	_, err = pool.Exec(ctx, `INSERT INTO folders (id, mailbox_id, name, special_use, total_count, unread_count) VALUES ($1, $2, 'INBOX', 'inbox', 2, 2)`,
+		folderID, mailboxID)
+	if err != nil {
+		t.Fatalf("seed folder: %v", err)
+	}
+	blobID := uuid.New()
+	_, err = pool.Exec(ctx, `INSERT INTO message_blobs (id, content_sha256, storage_driver, storage_path, size_bytes) VALUES ($1, $2, 'local', '/tmp/x', 10)`,
+		blobID, uuid.New().String()+uuid.New().String()[:32])
+	if err != nil {
+		t.Fatalf("seed blob: %v", err)
+	}
+
+	var msg1ID, msg2ID string
+	if err := pool.QueryRow(ctx, `
+		INSERT INTO email_messages (mailbox_id, folder_id, uid, blob_id, sender_address, recipient_addresses, size_bytes)
+		VALUES ($1, $2, 1, $3, 'sender@example.test', $4, 10) RETURNING id
+	`, mailboxID, folderID, blobID, []string{"user@" + domainName}).Scan(&msg1ID); err != nil {
+		t.Fatalf("seed message 1: %v", err)
+	}
+	if err := pool.QueryRow(ctx, `
+		INSERT INTO email_messages (mailbox_id, folder_id, uid, blob_id, sender_address, recipient_addresses, size_bytes)
+		VALUES ($1, $2, 2, $3, 'sender@example.test', $4, 10) RETURNING id
+	`, mailboxID, folderID, blobID, []string{"user@" + domainName}).Scan(&msg2ID); err != nil {
+		t.Fatalf("seed message 2: %v", err)
+	}
+
+	repo := NewExpungeRepository(pool)
+	if err := repo.Expunge(ctx, folderID.String(), []uint32{1}); err != nil {
+		t.Fatalf("Expunge: %v", err)
+	}
+
+	var expunged1 any
+	if err := pool.QueryRow(ctx, `SELECT expunged_at FROM email_messages WHERE id = $1`, msg1ID).Scan(&expunged1); err != nil {
+		t.Fatalf("query message 1: %v", err)
+	}
+	if expunged1 == nil {
+		t.Error("message 1 (uid=1, expunged) has expunged_at = NULL, want a timestamp")
+	}
+
+	var expunged2 any
+	if err := pool.QueryRow(ctx, `SELECT expunged_at FROM email_messages WHERE id = $1`, msg2ID).Scan(&expunged2); err != nil {
+		t.Fatalf("query message 2: %v", err)
+	}
+	if expunged2 != nil {
+		t.Error("message 2 (uid=2, not expunged) has a non-NULL expunged_at, want NULL")
+	}
+
+	var totalCount, unreadCount int
+	if err := pool.QueryRow(ctx, `SELECT total_count, unread_count FROM folders WHERE id = $1`, folderID).Scan(&totalCount, &unreadCount); err != nil {
+		t.Fatalf("query folder counters: %v", err)
+	}
+	if totalCount != 1 {
+		t.Errorf("total_count = %d, want 1 (started at 2, expunged 1)", totalCount)
+	}
+	if unreadCount != 1 {
+		t.Errorf("unread_count = %d, want 1 (expunged message had no \\Seen flag, so it counted as unread)", unreadCount)
+	}
+
+	var usedBytes int64
+	if err := pool.QueryRow(ctx, `SELECT used_bytes FROM mailboxes WHERE id = $1`, mailboxID).Scan(&usedBytes); err != nil {
+		t.Fatalf("query mailbox used_bytes: %v", err)
+	}
+	if usedBytes != 10 {
+		t.Errorf("used_bytes = %d, want 10 (started at 20, expunged message 1 with size_bytes=10)", usedBytes)
+	}
+}