From 0ec0698079e2cb3a49afefaf6fe3fac225e422e8 Mon Sep 17 00:00:00 2001 From: Lucas Rayan Date: Fri, 24 Jul 2026 09:40:47 -0300 Subject: [PATCH] #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 --- .../cmd/lambdamail-protocols/main.go | 10 +- .../application/port/copy_repository.go | 21 ++ .../application/port/expunge_repository.go | 16 + .../application/usecase/imap_session.go | 14 +- .../application/usecase/imap_session_test.go | 62 +++- .../internal/e2e/imap_core_e2e_test.go | 100 +++++++ .../postgres/copy_repository.go | 133 ++++++++ .../postgres/copy_repository_test.go | 110 +++++++ .../postgres/expunge_repository.go | 92 ++++++ .../postgres/expunge_repository_test.go | 96 ++++++ .../internal/presentation/imap/backend.go | 157 +++++++++- .../presentation/imap/backend_test.go | 163 +++++++++- .../internal/presentation/imap/search.go | 283 ++++++++++++++++++ .../internal/presentation/imap/search_test.go | 133 ++++++++ .../internal/presentation/imap/stubs.go | 9 - 15 files changed, 1378 insertions(+), 21 deletions(-) create mode 100644 services/protocols/internal/application/port/copy_repository.go create mode 100644 services/protocols/internal/application/port/expunge_repository.go create mode 100644 services/protocols/internal/infrastructure/postgres/copy_repository.go create mode 100644 services/protocols/internal/infrastructure/postgres/copy_repository_test.go create mode 100644 services/protocols/internal/infrastructure/postgres/expunge_repository.go create mode 100644 services/protocols/internal/infrastructure/postgres/expunge_repository_test.go create mode 100644 services/protocols/internal/presentation/imap/search.go create mode 100644 services/protocols/internal/presentation/imap/search_test.go diff --git a/services/protocols/cmd/lambdamail-protocols/main.go b/services/protocols/cmd/lambdamail-protocols/main.go index 4664a5a..92a36cf 100644 --- a/services/protocols/cmd/lambdamail-protocols/main.go +++ b/services/protocols/cmd/lambdamail-protocols/main.go @@ -8,6 +8,7 @@ import ( "net/http" "os" + "github.com/emersion/go-imap/v2" "github.com/emersion/go-imap/v2/imapserver" gosmtp "github.com/emersion/go-smtp" @@ -83,12 +84,19 @@ func main() { messageQuery := postgres.NewMessageQueryRepository(pool) flagRepo := postgres.NewFlagRepository(pool) blobReader := diskstorage.NewLocalDiskBlobReader(pool) - imapUseCase := usecase.NewImapSessionUseCase(authRepo, imapFolders, messageQuery, flagRepo, blobReader) + expungeRepo := postgres.NewExpungeRepository(pool) + copyRepo := postgres.NewCopyRepository(pool) + imapUseCase := usecase.NewImapSessionUseCase(authRepo, imapFolders, messageQuery, flagRepo, blobReader, expungeRepo, copyRepo) imapServer := imapserver.New(&imapserver.Options{ NewSession: func(c *imapserver.Conn) (imapserver.Session, *imapserver.GreetingData, error) { return imappresentation.NewSession(c, imapUseCase) }, + Caps: imap.CapSet{ + imap.CapIMAP4rev1: {}, + imap.CapMove: {}, + imap.CapUIDPlus: {}, + }, TLSConfig: &tls.Config{GetCertificate: certProvider.GetCertificate, MinVersion: tls.VersionTLS12}, InsecureAuth: false, }) diff --git a/services/protocols/internal/application/port/copy_repository.go b/services/protocols/internal/application/port/copy_repository.go new file mode 100644 index 0000000..654eb0d --- /dev/null +++ b/services/protocols/internal/application/port/copy_repository.go @@ -0,0 +1,21 @@ +package port + +import "context" + +// CopiedMessage records one message's UID before and after a COPY/MOVE. +type CopiedMessage struct { + SourceUID uint32 + DestUID uint32 +} + +// CopyRepository duplicates messages into another folder, preserving their +// blob reference (content-hash dedup - PLAN.md section 9) and flags. +type CopyRepository interface { + // CopyMessages copies each message identified by (sourceFolderID, uid) + // in uids into destFolderID, allocating a new UID in the destination + // per message, preserving flags, and incrementing the destination + // folder's total_count/unread_count and the destination mailbox's + // used_bytes accordingly. Returns one CopiedMessage per input uid, in + // the same order as uids. + CopyMessages(ctx context.Context, sourceFolderID string, uids []uint32, destFolderID string) ([]CopiedMessage, error) +} diff --git a/services/protocols/internal/application/port/expunge_repository.go b/services/protocols/internal/application/port/expunge_repository.go new file mode 100644 index 0000000..e5e1169 --- /dev/null +++ b/services/protocols/internal/application/port/expunge_repository.go @@ -0,0 +1,16 @@ +package port + +import "context" + +// ExpungeRepository permanently marks messages as expunged (soft delete - +// email_messages.expunged_at is set, never a hard DELETE). Blob ref_count +// is deliberately untouched here; physical blob garbage collection is a +// separate future concern. +type ExpungeRepository interface { + // Expunge marks each message identified by (folderID, uid) in uids as + // expunged, decrementing the folder's total_count (and unread_count for + // any message that lacked \Seen). The caller is responsible for having + // already determined that every UID in uids actually carries \Deleted - + // this method does not re-check that itself. + Expunge(ctx context.Context, folderID string, uids []uint32) error +} diff --git a/services/protocols/internal/application/usecase/imap_session.go b/services/protocols/internal/application/usecase/imap_session.go index c03e33c..31506a3 100644 --- a/services/protocols/internal/application/usecase/imap_session.go +++ b/services/protocols/internal/application/usecase/imap_session.go @@ -28,10 +28,12 @@ type ImapSessionUseCase struct { messages port.MessageQueryRepository flags port.FlagRepository blobs port.BlobReader + expunger port.ExpungeRepository + copier port.CopyRepository } -func NewImapSessionUseCase(auth port.AuthRepository, folders port.ImapFolderRepository, messages port.MessageQueryRepository, flags port.FlagRepository, blobs port.BlobReader) *ImapSessionUseCase { - return &ImapSessionUseCase{auth: auth, folders: folders, messages: messages, flags: flags, blobs: blobs} +func NewImapSessionUseCase(auth port.AuthRepository, folders port.ImapFolderRepository, messages port.MessageQueryRepository, flags port.FlagRepository, blobs port.BlobReader, expunger port.ExpungeRepository, copier port.CopyRepository) *ImapSessionUseCase { + return &ImapSessionUseCase{auth: auth, folders: folders, messages: messages, flags: flags, blobs: blobs, expunger: expunger, copier: copier} } func (uc *ImapSessionUseCase) Login(ctx context.Context, address, password string) (string, error) { @@ -78,3 +80,11 @@ func (uc *ImapSessionUseCase) ReadBlob(ctx context.Context, blobID uuid.UUID) ([ func (uc *ImapSessionUseCase) SetFlags(ctx context.Context, folderID string, uid uint32, op port.FlagOp, flags []string) error { return uc.flags.SetFlags(ctx, folderID, uid, op, flags) } + +func (uc *ImapSessionUseCase) Expunge(ctx context.Context, folderID string, uids []uint32) error { + return uc.expunger.Expunge(ctx, folderID, uids) +} + +func (uc *ImapSessionUseCase) CopyMessages(ctx context.Context, sourceFolderID string, uids []uint32, destFolderID string) ([]port.CopiedMessage, error) { + return uc.copier.CopyMessages(ctx, sourceFolderID, uids, destFolderID) +} diff --git a/services/protocols/internal/application/usecase/imap_session_test.go b/services/protocols/internal/application/usecase/imap_session_test.go index be73f85..2c4fc3f 100644 --- a/services/protocols/internal/application/usecase/imap_session_test.go +++ b/services/protocols/internal/application/usecase/imap_session_test.go @@ -32,7 +32,7 @@ func TestImapSessionUseCase_Login_SucceedsForCorrectPassword(t *testing.T) { auth := &fakeAuthRepository{byAddress: map[string]port.MailboxAuth{ "user@example.test": {ID: mailboxID, PasswordHash: testPasswordHash}, }} - uc := NewImapSessionUseCase(auth, nil, nil, nil, nil) + uc := NewImapSessionUseCase(auth, nil, nil, nil, nil, nil, nil) gotID, err := uc.Login(context.Background(), "user@example.test", testPassword) if err != nil { @@ -47,7 +47,7 @@ func TestImapSessionUseCase_Login_FailsForWrongPassword(t *testing.T) { auth := &fakeAuthRepository{byAddress: map[string]port.MailboxAuth{ "user@example.test": {ID: uuid.New(), PasswordHash: testPasswordHash}, }} - uc := NewImapSessionUseCase(auth, nil, nil, nil, nil) + uc := NewImapSessionUseCase(auth, nil, nil, nil, nil, nil, nil) _, err := uc.Login(context.Background(), "user@example.test", "wrong-password") if !errors.Is(err, ErrAuthFailed) { @@ -57,7 +57,7 @@ func TestImapSessionUseCase_Login_FailsForWrongPassword(t *testing.T) { func TestImapSessionUseCase_Login_FailsForUnknownAddress(t *testing.T) { auth := &fakeAuthRepository{byAddress: map[string]port.MailboxAuth{}} - uc := NewImapSessionUseCase(auth, nil, nil, nil, nil) + uc := NewImapSessionUseCase(auth, nil, nil, nil, nil, nil, nil) _, err := uc.Login(context.Background(), "nobody@example.test", "anything") if !errors.Is(err, ErrAuthFailed) { @@ -89,7 +89,7 @@ func TestImapSessionUseCase_SelectFolder_ReturnsRecordForExistingFolder(t *testi folders := &fakeImapFolderRepository{byName: map[string]port.ImapFolderRecord{ "INBOX": {ID: "folder-1", Name: "INBOX", UIDNext: 3, UIDValidity: 1000, NumMessages: 2}, }} - uc := NewImapSessionUseCase(nil, folders, nil, nil, nil) + uc := NewImapSessionUseCase(nil, folders, nil, nil, nil, nil, nil) rec, err := uc.SelectFolder(context.Background(), "mailbox-1", "INBOX") if err != nil { @@ -102,10 +102,62 @@ func TestImapSessionUseCase_SelectFolder_ReturnsRecordForExistingFolder(t *testi func TestImapSessionUseCase_SelectFolder_ReturnsErrNoSuchFolderWhenMissing(t *testing.T) { folders := &fakeImapFolderRepository{byName: map[string]port.ImapFolderRecord{}} - uc := NewImapSessionUseCase(nil, folders, nil, nil, nil) + uc := NewImapSessionUseCase(nil, folders, nil, nil, nil, nil, nil) _, err := uc.SelectFolder(context.Background(), "mailbox-1", "Nonexistent") if !errors.Is(err, ErrNoSuchFolder) { t.Fatalf("error = %v, want ErrNoSuchFolder", err) } } + +type fakeExpungeRepository struct { + calls []struct { + folderID string + uids []uint32 + } +} + +func (f *fakeExpungeRepository) Expunge(_ context.Context, folderID string, uids []uint32) error { + f.calls = append(f.calls, struct { + folderID string + uids []uint32 + }{folderID, uids}) + return nil +} + +type fakeCopyRepository struct { + result []port.CopiedMessage +} + +func (f *fakeCopyRepository) CopyMessages(_ context.Context, _ string, _ []uint32, _ string) ([]port.CopiedMessage, error) { + return f.result, nil +} + +func TestImapSessionUseCase_Expunge_DelegatesToRepository(t *testing.T) { + expunger := &fakeExpungeRepository{} + uc := NewImapSessionUseCase(nil, nil, nil, nil, nil, expunger, nil) + + err := uc.Expunge(context.Background(), "folder-1", []uint32{3, 7}) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(expunger.calls) != 1 || expunger.calls[0].folderID != "folder-1" { + t.Fatalf("calls = %+v, want one call with folderID=folder-1", expunger.calls) + } + if len(expunger.calls[0].uids) != 2 || expunger.calls[0].uids[0] != 3 || expunger.calls[0].uids[1] != 7 { + t.Errorf("uids = %v, want [3 7]", expunger.calls[0].uids) + } +} + +func TestImapSessionUseCase_CopyMessages_ReturnsRepositoryResult(t *testing.T) { + copier := &fakeCopyRepository{result: []port.CopiedMessage{{SourceUID: 3, DestUID: 1}, {SourceUID: 7, DestUID: 2}}} + uc := NewImapSessionUseCase(nil, nil, nil, nil, nil, nil, copier) + + got, err := uc.CopyMessages(context.Background(), "source-folder", []uint32{3, 7}, "dest-folder") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(got) != 2 || got[0].DestUID != 1 || got[1].DestUID != 2 { + t.Errorf("got = %+v, want [{3 1} {7 2}]", got) + } +} diff --git a/services/protocols/internal/e2e/imap_core_e2e_test.go b/services/protocols/internal/e2e/imap_core_e2e_test.go index 34d116f..a7270f8 100644 --- a/services/protocols/internal/e2e/imap_core_e2e_test.go +++ b/services/protocols/internal/e2e/imap_core_e2e_test.go @@ -108,11 +108,22 @@ func TestImapCoreEndToEnd(t *testing.T) { postgres.NewMessageQueryRepository(pool), postgres.NewFlagRepository(pool), diskstorage.NewLocalDiskBlobReader(pool), + postgres.NewExpungeRepository(pool), + postgres.NewCopyRepository(pool), ) server := imapserver.New(&imapserver.Options{ NewSession: func(c *imapserver.Conn) (imapserver.Session, *imapserver.GreetingData, error) { return imappresentation.NewSession(c, imapUC) }, + // Mirror main.go's Caps: without this, imapserver defaults to bare + // IMAP4rev1 and won't advertise MOVE/UIDPLUS, which the SEARCH/COPY/ + // EXPUNGE scenarios below (and the server-side COPYUID response) + // depend on the client library correctly attributing. + Caps: imap.CapSet{ + imap.CapIMAP4rev1: {}, + imap.CapMove: {}, + imap.CapUIDPlus: {}, + }, InsecureAuth: true, // no TLS in this test - InsecureAuth lets LOGIN proceed over plaintext }) ln, err := net.Listen("tcp", "127.0.0.1:0") @@ -196,4 +207,93 @@ func TestImapCoreEndToEnd(t *testing.T) { if !found { t.Errorf("flags after STORE = %v, want to include \\Seen", fetchAfterStore[0].Flags) } + + // --- SEARCH: the STORE above set \Seen on the one seeded message ------- + // + // SearchData.Count is only populated when the client requests an + // ESEARCH RETURN option (COUNT/MIN/MAX/ALL) - without SearchOptions the + // client sends a bare SEARCH command and the server replies with the + // classic "* SEARCH " response, which the client library only + // folds into SearchData.All (verified against imapclient/search.go's + // handleSearch, which only appends to cmd.data.All for that response + // form, vs handleESearch which is the only path that sets data.Count). + // So matches must be counted via AllSeqNums(), not Count. + searchData, err := client.Search(&imap.SearchCriteria{ + Flag: []imap.Flag{imap.FlagSeen}, + }, nil).Wait() + if err != nil { + t.Fatalf("SEARCH: %v", err) + } + if matched := searchData.AllSeqNums(); len(matched) != 1 { + t.Errorf("SEARCH \\Seen matched %v, want exactly 1 sequence number", matched) + } + + // --- COPY + EXPUNGE: seed a second folder the same way INBOX was seeded, + // copy the message into it, verify blob dedup accounting in Postgres, + // then expunge the (now \Deleted) INBOX original and confirm EXPUNGE + // only touched the selected folder. + archiveFolderID := "00000000-0000-0000-0000-000000000004" + if _, err := pool.Exec(ctx, `INSERT INTO folders (id, mailbox_id, name, special_use) VALUES ($1, $2, 'Archive', 'archive')`, archiveFolderID, mailboxID); err != nil { + t.Fatalf("seed Archive folder: %v", err) + } + + copyData, err := client.Copy(imap.UIDSetNum(1), "Archive").Wait() + if err != nil { + t.Fatalf("COPY: %v", err) + } + destUIDs, ok := copyData.DestUIDs.Nums() + if !ok || len(destUIDs) != 1 { + t.Fatalf("COPY DestUIDs = %v (ok=%v), want exactly 1 UID", destUIDs, ok) + } + + var inboxBlobID, archiveBlobID string + if err := pool.QueryRow(ctx, `SELECT blob_id FROM email_messages WHERE folder_id = $1 AND uid = 1 AND expunged_at IS NULL`, folderID).Scan(&inboxBlobID); err != nil { + t.Fatalf("query INBOX blob_id: %v", err) + } + var archiveCount int + if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM email_messages WHERE folder_id = $1 AND expunged_at IS NULL`, archiveFolderID).Scan(&archiveCount); err != nil { + t.Fatalf("count Archive messages: %v", err) + } + if archiveCount != 1 { + t.Errorf("Archive message count = %d, want 1", archiveCount) + } + if err := pool.QueryRow(ctx, `SELECT blob_id FROM email_messages WHERE folder_id = $1 AND expunged_at IS NULL`, archiveFolderID).Scan(&archiveBlobID); err != nil { + t.Fatalf("query Archive blob_id: %v", err) + } + if archiveBlobID != inboxBlobID { + t.Errorf("Archive blob_id = %s, want same blob as INBOX (%s)", archiveBlobID, inboxBlobID) + } + var refCount int + if err := pool.QueryRow(ctx, `SELECT ref_count FROM message_blobs WHERE id = $1`, inboxBlobID).Scan(&refCount); err != nil { + t.Fatalf("query blob ref_count: %v", err) + } + if refCount != 2 { + t.Errorf("message_blobs.ref_count = %d, want 2 after COPY", refCount) + } + + if _, err := client.Store(imap.UIDSetNum(1), &imap.StoreFlags{ + Op: imap.StoreFlagsAdd, Flags: []imap.Flag{imap.FlagDeleted}, + }, nil).Collect(); err != nil { + t.Fatalf("STORE \\Deleted: %v", err) + } + + if _, err := client.Expunge().Collect(); err != nil { + t.Fatalf("EXPUNGE: %v", err) + } + + var inboxCount int + if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM email_messages WHERE folder_id = $1 AND expunged_at IS NULL`, folderID).Scan(&inboxCount); err != nil { + t.Fatalf("count INBOX messages after EXPUNGE: %v", err) + } + if inboxCount != 0 { + t.Errorf("INBOX message count after EXPUNGE = %d, want 0", inboxCount) + } + + var archiveCountAfter int + if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM email_messages WHERE folder_id = $1 AND expunged_at IS NULL`, archiveFolderID).Scan(&archiveCountAfter); err != nil { + t.Fatalf("count Archive messages after EXPUNGE: %v", err) + } + if archiveCountAfter != 1 { + t.Errorf("Archive message count after INBOX EXPUNGE = %d, want 1 (untouched)", archiveCountAfter) + } } diff --git a/services/protocols/internal/infrastructure/postgres/copy_repository.go b/services/protocols/internal/infrastructure/postgres/copy_repository.go new file mode 100644 index 0000000..7e0bcb2 --- /dev/null +++ b/services/protocols/internal/infrastructure/postgres/copy_repository.go @@ -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 +} diff --git a/services/protocols/internal/infrastructure/postgres/copy_repository_test.go b/services/protocols/internal/infrastructure/postgres/copy_repository_test.go new file mode 100644 index 0000000..ea397db --- /dev/null +++ b/services/protocols/internal/infrastructure/postgres/copy_repository_test.go @@ -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 +} diff --git a/services/protocols/internal/infrastructure/postgres/expunge_repository.go b/services/protocols/internal/infrastructure/postgres/expunge_repository.go new file mode 100644 index 0000000..cb61fd2 --- /dev/null +++ b/services/protocols/internal/infrastructure/postgres/expunge_repository.go @@ -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 +} diff --git a/services/protocols/internal/infrastructure/postgres/expunge_repository_test.go b/services/protocols/internal/infrastructure/postgres/expunge_repository_test.go new file mode 100644 index 0000000..88303bd --- /dev/null +++ b/services/protocols/internal/infrastructure/postgres/expunge_repository_test.go @@ -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) + } +} diff --git a/services/protocols/internal/presentation/imap/backend.go b/services/protocols/internal/presentation/imap/backend.go index f4c3e8c..68941f5 100644 --- a/services/protocols/internal/presentation/imap/backend.go +++ b/services/protocols/internal/presentation/imap/backend.go @@ -123,7 +123,7 @@ func matchFolders(recs []port.ImapFolderRecord, ref string, patterns []string) [ // numSet contains that message (matching imapmemserver's own reference // pattern of iterating known messages and calling NumSet.Contains, rather // than trying to resolve "*" ranges ourselves). -func (s *session) numSetIter(numSet imap.NumSet, isUID bool, messages []port.MessageRecord, f func(seqNum uint32, msg port.MessageRecord)) { +func (s *session) numSetIter(numSet imap.NumSet, messages []port.MessageRecord, f func(seqNum uint32, msg port.MessageRecord)) { for i, msg := range messages { seqNum := uint32(i) + 1 var contains bool @@ -149,7 +149,7 @@ func (s *session) Fetch(w *imapserver.FetchWriter, numSet imap.NumSet, options * } var fetchErr error - s.numSetIter(numSet, options.UID, messages, func(seqNum uint32, msg port.MessageRecord) { + s.numSetIter(numSet, messages, func(seqNum uint32, msg port.MessageRecord) { if fetchErr != nil { return } @@ -207,6 +207,157 @@ func (s *session) writeFetchResponse(w *imapserver.FetchResponseWriter, msg port return w.Close() } +// expungeUIDs marks messages expunged: when requireDeleted is true, every +// message in the selected folder carrying \Deleted (used by EXPUNGE, per RFC +// 3501); when false, every message matched by uids regardless of flags (used +// by MOVE, per RFC 6851 - "MOVE is not required to use the \Deleted flag to +// mark messages", precisely so a client doesn't have to flag-then-expunge). +// It tells the use case to mark the matched UIDs expunged, and returns their +// sequence numbers in descending order (so callers can report EXPUNGE +// responses without a removal invalidating the sequence numbers of messages +// reported afterward - matches go-imap/v2's own imapmemserver reference +// behavior). Shared by Expunge and Move, which each write their own EXPUNGE +// responses against their own writer type (*imapserver.ExpungeWriter vs +// *imapserver.MoveWriter): both types embed an unexported *imapserver.Conn +// field set only by the imapserver package itself, so neither can be +// constructed outside a live connection, and Move cannot simply delegate to +// Expunge's full method. +func (s *session) expungeUIDs(uids *imap.UIDSet, requireDeleted bool) ([]uint32, error) { + if s.selectedFolder == nil { + return nil, &imap.Error{Type: imap.StatusResponseTypeNo, Text: "No mailbox selected"} + } + messages, err := s.useCase.FetchMessages(context.Background(), s.selectedFolder.ID) + if err != nil { + return nil, toIMAPError(err) + } + + var toExpunge []uint32 + var seqNums []uint32 + for i, msg := range messages { + seqNum := uint32(i) + 1 + if uids != nil && !uids.Contains(imap.UID(msg.UID)) { + continue + } + if requireDeleted { + hasDeleted := false + for _, f := range msg.Flags { + if imap.Flag(f) == imap.FlagDeleted { + hasDeleted = true + break + } + } + if !hasDeleted { + continue + } + } + toExpunge = append(toExpunge, msg.UID) + seqNums = append(seqNums, seqNum) + } + if len(toExpunge) == 0 { + return nil, nil + } + + if err := s.useCase.Expunge(context.Background(), s.selectedFolder.ID, toExpunge); err != nil { + return nil, toIMAPError(err) + } + + // Reverse in place: descending sequence-number order. + for i, j := 0, len(seqNums)-1; i < j; i, j = i+1, j-1 { + seqNums[i], seqNums[j] = seqNums[j], seqNums[i] + } + return seqNums, nil +} + +func (s *session) Expunge(w *imapserver.ExpungeWriter, uids *imap.UIDSet) error { + seqNums, err := s.expungeUIDs(uids, true) + if err != nil { + return err + } + // w is nil in unit tests that only assert the use-case call, since + // *imapserver.ExpungeWriter has no exported constructor and can only be + // built by a live *imapserver.Conn (same constraint as *ListWriter - + // see matchFolders' doc comment above). The real imapserver package + // never passes a nil writer. + if w == nil { + return nil + } + for _, seqNum := range seqNums { + if err := w.WriteExpunge(seqNum); err != nil { + return err + } + } + return nil +} + +func (s *session) Copy(numSet imap.NumSet, dest string) (*imap.CopyData, error) { + if s.selectedFolder == nil { + return nil, &imap.Error{Type: imap.StatusResponseTypeNo, Text: "No mailbox selected"} + } + destFolder, err := s.useCase.SelectFolder(context.Background(), s.mailboxID, dest) + if err != nil { + return nil, &imap.Error{Type: imap.StatusResponseTypeNo, Code: imap.ResponseCodeTryCreate, Text: "No such destination mailbox"} + } + + messages, err := s.useCase.FetchMessages(context.Background(), s.selectedFolder.ID) + if err != nil { + return nil, toIMAPError(err) + } + var uids []uint32 + s.numSetIter(numSet, messages, func(_ uint32, msg port.MessageRecord) { + uids = append(uids, msg.UID) + }) + + copied, err := s.useCase.CopyMessages(context.Background(), s.selectedFolder.ID, uids, destFolder.ID) + if err != nil { + return nil, toIMAPError(err) + } + + var sourceUIDs, destUIDs imap.UIDSet + for _, c := range copied { + sourceUIDs.AddNum(imap.UID(c.SourceUID)) + destUIDs.AddNum(imap.UID(c.DestUID)) + } + return &imap.CopyData{UIDValidity: destFolder.UIDValidity, SourceUIDs: sourceUIDs, DestUIDs: destUIDs}, nil +} + +// Move copies numSet into dest (reusing Copy's logic), reports the COPYUID +// data, then expunges the just-copied source messages (RFC 6851: MOVE is +// COPY + EXPUNGE-of-the-moved-messages-only, atomically from the client's +// point of view). It deliberately does NOT call s.Expunge: that method +// writes against *imapserver.ExpungeWriter, a different concrete type from +// the *imapserver.MoveWriter this method receives, even though both expose a +// method named WriteExpunge - so Move writes its own EXPUNGE responses +// through its own writer, sharing only the expungeUIDs marking logic. +func (s *session) Move(w *imapserver.MoveWriter, numSet imap.NumSet, dest string) error { + copyData, err := s.Copy(numSet, dest) + if err != nil { + return err + } + if w != nil { + if err := w.WriteCopyData(copyData); err != nil { + return err + } + } + + // MOVE does not require \Deleted (RFC 6851): expunge exactly the + // UIDs Copy just copied, regardless of their flags. + seqNums, err := s.expungeUIDs(©Data.SourceUIDs, false) + if err != nil { + return err + } + // w is nil in unit tests that only assert the use-case call, mirroring + // Expunge's nil-writer accommodation above. + if w == nil { + return nil + } + for _, seqNum := range seqNums { + if err := w.WriteExpunge(seqNum); err != nil { + return err + } + } + return nil +} + func (s *session) Store(w *imapserver.FetchWriter, numSet imap.NumSet, flags *imap.StoreFlags, options *imap.StoreOptions) error { if s.selectedFolder == nil { return &imap.Error{Type: imap.StatusResponseTypeNo, Text: "No mailbox selected"} @@ -236,7 +387,7 @@ func (s *session) Store(w *imapserver.FetchWriter, numSet imap.NumSet, flags *im // test against, and Nums() returns (nil, false) for them. This also // naturally supports plain (non-UID) STORE via the SeqSet case. var storeErr error - s.numSetIter(numSet, true, messages, func(_ uint32, msg port.MessageRecord) { + s.numSetIter(numSet, messages, func(_ uint32, msg port.MessageRecord) { if storeErr != nil { return } diff --git a/services/protocols/internal/presentation/imap/backend_test.go b/services/protocols/internal/presentation/imap/backend_test.go index b489429..03cdd67 100644 --- a/services/protocols/internal/presentation/imap/backend_test.go +++ b/services/protocols/internal/presentation/imap/backend_test.go @@ -83,6 +83,32 @@ func (f *fakeBlobReader) ReadByID(_ context.Context, blobID uuid.UUID) ([]byte, return f.content[blobID], nil } +// fakeExpungeRepository/fakeCopyRepository duplicate the fakes from +// internal/application/usecase/imap_session_test.go - Go test files in +// different packages can't share unexported test helpers. +type fakeExpungeRepository struct { + calls []struct { + folderID string + uids []uint32 + } +} + +func (f *fakeExpungeRepository) Expunge(_ context.Context, folderID string, uids []uint32) error { + f.calls = append(f.calls, struct { + folderID string + uids []uint32 + }{folderID, uids}) + return nil +} + +type fakeCopyRepository struct { + result []port.CopiedMessage +} + +func (f *fakeCopyRepository) CopyMessages(_ context.Context, _ string, _ []uint32, _ string) ([]port.CopiedMessage, error) { + return f.result, nil +} + // testPasswordHash/testPassword duplicate the constants from // internal/application/usecase/imap_session_test.go - a real Argon2id PHC // hash for "correct horse battery staple", generated once with @@ -94,7 +120,24 @@ const ( ) func newTestSession(auth *fakeAuthRepository, folders *fakeImapFolderRepository, messages *fakeMessageQueryRepository, flags *fakeFlagRepository, blobs *fakeBlobReader) imapserver.Session { - uc := usecase.NewImapSessionUseCase(auth, folders, messages, flags, blobs) + // expunger/copier are nil here: none of the tests using this helper + // exercise Expunge/CopyMessages, and NewImapSessionUseCase only stores + // these interfaces without dereferencing them at construction time. + // Tests that DO exercise Expunge/Copy use + // newTestSessionWithExpungeAndCopy below instead. + return newTestSessionWithExpungeAndCopy(auth, folders, messages, flags, blobs, nil, nil) +} + +func newTestSessionWithExpungeAndCopy(auth *fakeAuthRepository, folders *fakeImapFolderRepository, messages *fakeMessageQueryRepository, flags *fakeFlagRepository, blobs *fakeBlobReader, expunger *fakeExpungeRepository, copier *fakeCopyRepository) imapserver.Session { + var expungeRepo port.ExpungeRepository + if expunger != nil { + expungeRepo = expunger + } + var copyRepo port.CopyRepository + if copier != nil { + copyRepo = copier + } + uc := usecase.NewImapSessionUseCase(auth, folders, messages, flags, blobs, expungeRepo, copyRepo) sess, _, err := NewSession(nil, uc) if err != nil { panic(err) @@ -403,6 +446,124 @@ func TestSession_List_UsesFolderRepository(t *testing.T) { } } +func TestSession_Expunge_CallsUseCaseWithDeletedUIDsOnly(t *testing.T) { + mailboxID := uuid.New() + auth := &fakeAuthRepository{byAddress: map[string]port.MailboxAuth{ + "user@example.test": {ID: mailboxID, PasswordHash: testPasswordHash}, + }} + folders := &fakeImapFolderRepository{byName: map[string]port.ImapFolderRecord{"INBOX": {ID: "folder-1"}}} + messages := &fakeMessageQueryRepository{messages: []port.MessageRecord{ + {UID: 1, Flags: []string{"\\Deleted"}}, + {UID: 2, Flags: nil}, + {UID: 3, Flags: []string{"\\Deleted", "\\Seen"}}, + }} + expunger := &fakeExpungeRepository{} + sess := newTestSessionWithExpungeAndCopy(auth, folders, messages, &fakeFlagRepository{}, &fakeBlobReader{}, expunger, &fakeCopyRepository{}) + _ = sess.Login("user@example.test", testPassword) + if _, err := sess.Select("INBOX", &imap.SelectOptions{}); err != nil { + t.Fatalf("Select: %v", err) + } + + if err := sess.Expunge(nil, nil); err != nil { + t.Fatalf("Expunge: %v", err) + } + if len(expunger.calls) != 1 { + t.Fatalf("expunger called %d times, want 1", len(expunger.calls)) + } + uids := expunger.calls[0].uids + if len(uids) != 2 || uids[0] != 1 || uids[1] != 3 { + t.Errorf("expunged uids = %v, want [1 3] (only \\Deleted messages)", uids) + } +} + +func TestSession_Copy_CallsUseCaseAndBuildsCopyData(t *testing.T) { + mailboxID := uuid.New() + auth := &fakeAuthRepository{byAddress: map[string]port.MailboxAuth{ + "user@example.test": {ID: mailboxID, PasswordHash: testPasswordHash}, + }} + folders := &fakeImapFolderRepository{byName: map[string]port.ImapFolderRecord{ + "INBOX": {ID: "folder-1"}, + "Archive": {ID: "folder-2", UIDValidity: 42}, + }} + messages := &fakeMessageQueryRepository{messages: []port.MessageRecord{{UID: 5}}} + copier := &fakeCopyRepository{result: []port.CopiedMessage{{SourceUID: 5, DestUID: 1}}} + sess := newTestSessionWithExpungeAndCopy(auth, folders, messages, &fakeFlagRepository{}, &fakeBlobReader{}, &fakeExpungeRepository{}, copier) + _ = sess.Login("user@example.test", testPassword) + _, _ = sess.Select("INBOX", &imap.SelectOptions{}) + + data, err := sess.Copy(imap.UIDSetNum(5), "Archive") + if err != nil { + t.Fatalf("Copy: %v", err) + } + if data.UIDValidity != 42 { + t.Errorf("UIDValidity = %d, want 42", data.UIDValidity) + } + srcNums, _ := data.SourceUIDs.Nums() + destNums, _ := data.DestUIDs.Nums() + if len(srcNums) != 1 || srcNums[0] != 5 || len(destNums) != 1 || destNums[0] != 1 { + t.Errorf("SourceUIDs=%v DestUIDs=%v, want [5]/[1]", srcNums, destNums) + } +} + +func TestSession_Copy_FailsForUnknownDestination(t *testing.T) { + mailboxID := uuid.New() + auth := &fakeAuthRepository{byAddress: map[string]port.MailboxAuth{ + "user@example.test": {ID: mailboxID, PasswordHash: testPasswordHash}, + }} + folders := &fakeImapFolderRepository{byName: map[string]port.ImapFolderRecord{"INBOX": {ID: "folder-1"}}} + sess := newTestSessionWithExpungeAndCopy(auth, folders, &fakeMessageQueryRepository{}, &fakeFlagRepository{}, &fakeBlobReader{}, &fakeExpungeRepository{}, &fakeCopyRepository{}) + _ = sess.Login("user@example.test", testPassword) + _, _ = sess.Select("INBOX", &imap.SelectOptions{}) + + _, err := sess.Copy(imap.UIDSetNum(5), "Nonexistent") + if err == nil { + t.Fatal("expected an error copying to a nonexistent folder, got nil") + } +} + +// TestSession_Move_ExpungesCopiedMessageEvenWithoutDeletedFlag proves the +// RFC 6851 fix: MOVE must expunge exactly the UIDs it just copied, +// regardless of whether they carry \Deleted. Before this fix, Move called +// the shared expungeUIDs helper in its \Deleted-requiring mode (the one +// correct for EXPUNGE), so a message with no \Deleted flag would be copied +// into the destination but never removed from the source folder - a silent +// duplication instead of a move. +func TestSession_Move_ExpungesCopiedMessageEvenWithoutDeletedFlag(t *testing.T) { + mailboxID := uuid.New() + auth := &fakeAuthRepository{byAddress: map[string]port.MailboxAuth{ + "user@example.test": {ID: mailboxID, PasswordHash: testPasswordHash}, + }} + folders := &fakeImapFolderRepository{byName: map[string]port.ImapFolderRecord{ + "INBOX": {ID: "folder-1"}, + "Archive": {ID: "folder-2", UIDValidity: 42}, + }} + // UID 5 has no \Deleted flag - MOVE must still expunge it from the + // source folder, unlike EXPUNGE which would skip it. + messages := &fakeMessageQueryRepository{messages: []port.MessageRecord{{UID: 5, Flags: nil}}} + expunger := &fakeExpungeRepository{} + copier := &fakeCopyRepository{result: []port.CopiedMessage{{SourceUID: 5, DestUID: 1}}} + sess := newTestSessionWithExpungeAndCopy(auth, folders, messages, &fakeFlagRepository{}, &fakeBlobReader{}, expunger, copier) + _ = sess.Login("user@example.test", testPassword) + if _, err := sess.Select("INBOX", &imap.SelectOptions{}); err != nil { + t.Fatalf("Select: %v", err) + } + + // w is nil here for the same reason Expunge's tests pass a nil writer: + // *imapserver.MoveWriter has no exported constructor and can only be + // built by a live *imapserver.Conn. + if err := sess.(*session).Move(nil, imap.UIDSetNum(5), "Archive"); err != nil { + t.Fatalf("Move: %v", err) + } + + if len(expunger.calls) != 1 { + t.Fatalf("expunger called %d times, want 1", len(expunger.calls)) + } + uids := expunger.calls[0].uids + if len(uids) != 1 || uids[0] != 5 { + t.Errorf("expunged uids = %v, want [5] (moved even though it has no \\Deleted flag)", uids) + } +} + func TestStubMethods_ReturnNotYetImplementedError(t *testing.T) { mailboxID := uuid.New() auth := &fakeAuthRepository{byAddress: map[string]port.MailboxAuth{ diff --git a/services/protocols/internal/presentation/imap/search.go b/services/protocols/internal/presentation/imap/search.go new file mode 100644 index 0000000..1f9a774 --- /dev/null +++ b/services/protocols/internal/presentation/imap/search.go @@ -0,0 +1,283 @@ +// Package imappresentation: this file implements IMAP SEARCH (RFC 3501 +// section 6.4.4). Criteria matching is a pure function operating on data +// already available via the application layer (FetchMessages, ReadBlob) - +// it intentionally lives here, not in the use case, because imap. +// SearchCriteria is a go-imap/v2 type and the application layer must stay +// framework-free (see this plan's Global Constraints). +package imappresentation + +import ( + "bufio" + "bytes" + "context" + "io" + "strings" + "time" + + "github.com/emersion/go-imap/v2" + "github.com/emersion/go-imap/v2/imapserver" + gomessage "github.com/emersion/go-message" + "github.com/emersion/go-message/mail" + "github.com/emersion/go-message/textproto" +) + +func (s *session) Search(numKind imapserver.NumKind, criteria *imap.SearchCriteria, _ *imap.SearchOptions) (*imap.SearchData, error) { + if s.selectedFolder == nil { + return nil, &imap.Error{Type: imap.StatusResponseTypeNo, Text: "No mailbox selected"} + } + messages, err := s.useCase.FetchMessages(context.Background(), s.selectedFolder.ID) + if err != nil { + return nil, toIMAPError(err) + } + + data := &imap.SearchData{} + var seqSet imap.SeqSet + var uidSet imap.UIDSet + + for i, msg := range messages { + seqNum := uint32(i) + 1 + flags := make([]imap.Flag, len(msg.Flags)) + for j, f := range msg.Flags { + flags[j] = imap.Flag(f) + } + + blobID := msg.BlobID + loadRaw := func() ([]byte, error) { + return s.useCase.ReadBlob(context.Background(), blobID) + } + + if !matchCriteria(seqNum, msg.UID, msg.ReceivedAt, msg.SizeBytes, flags, criteria, loadRaw) { + continue + } + + seqSet.AddNum(seqNum) + uidSet.AddNum(imap.UID(msg.UID)) + var num uint32 + if numKind == imapserver.NumKindUID { + num = msg.UID + } else { + num = seqNum + } + if data.Min == 0 || num < data.Min { + data.Min = num + } + if num > data.Max { + data.Max = num + } + data.Count++ + } + + if numKind == imapserver.NumKindUID { + data.All = uidSet + } else { + data.All = seqSet + } + return data, nil +} + +// matchCriteria mirrors go-imap/v2's own imapmemserver reference matching +// logic (imapserver/imapmemserver/message.go's (*message).search), adapted +// to lazily load raw message bytes via loadRaw only when a criterion +// actually needs them (Header/Body/Text/SentSince/SentBefore), since this +// codebase is Postgres-backed with blobs fetched on demand rather than +// always held in memory like imapmemserver's reference message buffer. +func matchCriteria(seqNum, uid uint32, receivedAt time.Time, sizeBytes int64, flags []imap.Flag, criteria *imap.SearchCriteria, loadRaw func() ([]byte, error)) bool { + for _, seqSet := range criteria.SeqNum { + if seqNum == 0 || !seqSet.Contains(seqNum) { + return false + } + } + for _, uidSet := range criteria.UID { + if !uidSet.Contains(imap.UID(uid)) { + return false + } + } + if !matchDate(receivedAt, criteria.Since, criteria.Before) { + return false + } + + flagSet := make(map[imap.Flag]struct{}, len(flags)) + for _, f := range flags { + flagSet[canonicalFlag(f)] = struct{}{} + } + for _, flag := range criteria.Flag { + if _, ok := flagSet[canonicalFlag(flag)]; !ok { + return false + } + } + for _, flag := range criteria.NotFlag { + if _, ok := flagSet[canonicalFlag(flag)]; ok { + return false + } + } + + if criteria.Larger != 0 && sizeBytes <= criteria.Larger { + return false + } + if criteria.Smaller != 0 && sizeBytes >= criteria.Smaller { + return false + } + + needsRaw := len(criteria.Header) > 0 || len(criteria.Body) > 0 || len(criteria.Text) > 0 || + !criteria.SentSince.IsZero() || !criteria.SentBefore.IsZero() + var header textproto.Header + var raw []byte + if needsRaw { + var err error + raw, err = loadRaw() + if err != nil { + return false + } + header, err = textproto.ReadHeader(bufio.NewReader(bytes.NewReader(raw))) + if err != nil { + return false + } + } + + for _, fieldCriteria := range criteria.Header { + if !matchHeaderField(header, fieldCriteria.Key, fieldCriteria.Value) { + return false + } + } + + if !criteria.SentSince.IsZero() || !criteria.SentBefore.IsZero() { + t, err := parseDateHeader(header) + if err != nil || !matchDate(t, criteria.SentSince, criteria.SentBefore) { + return false + } + } + + for _, text := range criteria.Text { + if !matchEntity(raw, text, true) { + return false + } + } + for _, body := range criteria.Body { + if !matchEntity(raw, body, false) { + return false + } + } + + for _, not := range criteria.Not { + if matchCriteria(seqNum, uid, receivedAt, sizeBytes, flags, ¬, loadRaw) { + return false + } + } + for _, or := range criteria.Or { + if !matchCriteria(seqNum, uid, receivedAt, sizeBytes, flags, &or[0], loadRaw) && + !matchCriteria(seqNum, uid, receivedAt, sizeBytes, flags, &or[1], loadRaw) { + return false + } + } + + return true +} + +// canonicalFlag normalizes flag comparisons so e.g. "\Seen" and "\seen" are +// treated as equal, matching IMAP's case-insensitive flag semantics. +func canonicalFlag(f imap.Flag) imap.Flag { + return imap.Flag(strings.ToLower(string(f))) +} + +func matchDate(t, since, before time.Time) bool { + // RFC 3501 explicitly requires zone-unaware date (not time) comparison. + t = time.Date(t.Year(), t.Month(), t.Day(), 0, 0, 0, 0, time.UTC) + if !since.IsZero() && t.Before(since) { + return false + } + if !before.IsZero() && !t.Before(before) { + return false + } + return true +} + +func matchHeaderField(header textproto.Header, key, pattern string) bool { + fields := header.FieldsByKey(key) + if pattern == "" { + return fields.Len() > 0 + } + pattern = strings.ToLower(pattern) + for fields.Next() { + if strings.Contains(strings.ToLower(fields.Value()), pattern) { + return true + } + } + return false +} + +// parseDateHeader parses the message's Date: header field. +// +// The real API (verified against github.com/emersion/go-message@installed +// version, and against how go-imap/v2's own imapserver/imapmemserver +// reference implementation does this in message.go's (*message).search: +// `header := mail.Header{msg.reader().Header}` then `header.Date()`) is +// *not* a type assertion on Header.Fields() - that sketch does not compile, +// since the iterator returned by Fields() has no Date() method. Instead, +// github.com/emersion/go-message/mail.Header wraps a +// github.com/emersion/go-message.Header (which itself wraps +// textproto.Header) and exposes a `Date() (time.Time, error)` method that +// does the real RFC 5322 date parsing. So a textproto.Header must be +// wrapped twice - once into gomessage.Header, once into mail.Header - to +// reach that method. +func parseDateHeader(header textproto.Header) (time.Time, error) { + h := mail.Header{Header: gomessage.Header{Header: header}} + return h.Date() +} + +// matchEntity re-parses raw as a MIME entity and checks whether pattern +// (case-insensitively) appears in the body (and, if includeHeader is true, +// also the header fields) - mirroring imapmemserver's matchEntity, which +// walks multipart parts recursively via gomessage.Entity rather than doing +// a flat byte search over raw, so that HTML/text alternative parts and +// nested MIME structures are still searched correctly. +func matchEntity(raw []byte, pattern string, includeHeader bool) bool { + if pattern == "" { + return true + } + entity, err := gomessage.Read(bytes.NewReader(raw)) + if err != nil && entity == nil { + return false + } + return matchEntityRecursive(entity, strings.ToLower(pattern), includeHeader) +} + +func matchEntityRecursive(e *gomessage.Entity, patternLower string, includeHeader bool) bool { + if e == nil { + return false + } + if includeHeader && matchHeaderFieldsLower(e.Header.Fields(), patternLower) { + return true + } + + if mr := e.MultipartReader(); mr != nil { + for { + part, err := mr.NextPart() + if err == io.EOF { + break + } + if err != nil { + break + } + if matchEntityRecursive(part, patternLower, includeHeader) { + return true + } + } + return false + } + + body, err := io.ReadAll(e.Body) + if err != nil { + return false + } + return strings.Contains(strings.ToLower(string(body)), patternLower) +} + +func matchHeaderFieldsLower(fields gomessage.HeaderFields, patternLower string) bool { + for fields.Next() { + v, _ := fields.Text() + if strings.Contains(strings.ToLower(v), patternLower) { + return true + } + } + return false +} diff --git a/services/protocols/internal/presentation/imap/search_test.go b/services/protocols/internal/presentation/imap/search_test.go new file mode 100644 index 0000000..1ee07f1 --- /dev/null +++ b/services/protocols/internal/presentation/imap/search_test.go @@ -0,0 +1,133 @@ +package imappresentation + +import ( + "testing" + "time" + + "github.com/emersion/go-imap/v2" + "github.com/emersion/go-imap/v2/imapserver" + "github.com/google/uuid" + + "lambdamail/protocols/internal/application/port" +) + +func TestSession_Search_MatchesByFlag(t *testing.T) { + mailboxID := uuid.New() + auth := &fakeAuthRepository{byAddress: map[string]port.MailboxAuth{ + "user@example.test": {ID: mailboxID, PasswordHash: testPasswordHash}, + }} + folders := &fakeImapFolderRepository{byName: map[string]port.ImapFolderRecord{ + "INBOX": {ID: "folder-1"}, + }} + messages := &fakeMessageQueryRepository{messages: []port.MessageRecord{ + {UID: 1, ReceivedAt: time.Now(), Flags: []string{"\\Seen"}}, + {UID: 2, ReceivedAt: time.Now(), Flags: nil}, + {UID: 3, ReceivedAt: time.Now(), Flags: []string{"\\Seen", "\\Flagged"}}, + }} + sess := newTestSession(auth, folders, messages, &fakeFlagRepository{}, &fakeBlobReader{}) + _ = sess.Login("user@example.test", testPassword) + if _, err := sess.Select("INBOX", &imap.SelectOptions{}); err != nil { + t.Fatalf("Select: %v", err) + } + + data, err := sess.Search(imapserver.NumKindUID, &imap.SearchCriteria{Flag: []imap.Flag{imap.FlagSeen}}, &imap.SearchOptions{}) + if err != nil { + t.Fatalf("Search: %v", err) + } + uidSet, ok := data.All.(imap.UIDSet) + if !ok { + t.Fatalf("data.All is %T, want imap.UIDSet", data.All) + } + nums, ok := uidSet.Nums() + if !ok || len(nums) != 2 || nums[0] != 1 || nums[1] != 3 { + t.Errorf("matched UIDs = %v (ok=%v), want [1 3]", nums, ok) + } +} + +func TestSession_Search_MatchesByUIDSetIntersection(t *testing.T) { + mailboxID := uuid.New() + auth := &fakeAuthRepository{byAddress: map[string]port.MailboxAuth{ + "user@example.test": {ID: mailboxID, PasswordHash: testPasswordHash}, + }} + folders := &fakeImapFolderRepository{byName: map[string]port.ImapFolderRecord{"INBOX": {ID: "folder-1"}}} + messages := &fakeMessageQueryRepository{messages: []port.MessageRecord{ + {UID: 1, ReceivedAt: time.Now()}, + {UID: 2, ReceivedAt: time.Now()}, + {UID: 3, ReceivedAt: time.Now()}, + }} + sess := newTestSession(auth, folders, messages, &fakeFlagRepository{}, &fakeBlobReader{}) + _ = sess.Login("user@example.test", testPassword) + _, _ = sess.Select("INBOX", &imap.SelectOptions{}) + + var uidFilter imap.UIDSet + uidFilter.AddRange(2, 0) // "2:*" + data, err := sess.Search(imapserver.NumKindUID, &imap.SearchCriteria{UID: []imap.UIDSet{uidFilter}}, &imap.SearchOptions{}) + if err != nil { + t.Fatalf("Search: %v", err) + } + uidSet := data.All.(imap.UIDSet) + nums, _ := uidSet.Nums() + if len(nums) != 2 || nums[0] != 2 || nums[1] != 3 { + t.Errorf("matched UIDs = %v, want [2 3]", nums) + } +} + +func TestSession_Search_MatchesByHeaderField(t *testing.T) { + mailboxID := uuid.New() + auth := &fakeAuthRepository{byAddress: map[string]port.MailboxAuth{ + "user@example.test": {ID: mailboxID, PasswordHash: testPasswordHash}, + }} + folders := &fakeImapFolderRepository{byName: map[string]port.ImapFolderRecord{"INBOX": {ID: "folder-1"}}} + blobA := uuid.New() + blobB := uuid.New() + messages := &fakeMessageQueryRepository{messages: []port.MessageRecord{ + {UID: 1, BlobID: blobA, ReceivedAt: time.Now()}, + {UID: 2, BlobID: blobB, ReceivedAt: time.Now()}, + }} + blobs := &fakeBlobReader{content: map[uuid.UUID][]byte{ + blobA: []byte("Subject: hello world\r\n\r\nbody"), + blobB: []byte("Subject: something else\r\n\r\nbody"), + }} + sess := newTestSession(auth, folders, messages, &fakeFlagRepository{}, blobs) + _ = sess.Login("user@example.test", testPassword) + _, _ = sess.Select("INBOX", &imap.SelectOptions{}) + + data, err := sess.Search(imapserver.NumKindUID, &imap.SearchCriteria{ + Header: []imap.SearchCriteriaHeaderField{{Key: "Subject", Value: "hello"}}, + }, &imap.SearchOptions{}) + if err != nil { + t.Fatalf("Search: %v", err) + } + uidSet := data.All.(imap.UIDSet) + nums, _ := uidSet.Nums() + if len(nums) != 1 || nums[0] != 1 { + t.Errorf("matched UIDs = %v, want [1] (only the message with \"hello\" in Subject)", nums) + } +} + +func TestSession_Search_NotCriteriaExcludesMatches(t *testing.T) { + mailboxID := uuid.New() + auth := &fakeAuthRepository{byAddress: map[string]port.MailboxAuth{ + "user@example.test": {ID: mailboxID, PasswordHash: testPasswordHash}, + }} + folders := &fakeImapFolderRepository{byName: map[string]port.ImapFolderRecord{"INBOX": {ID: "folder-1"}}} + messages := &fakeMessageQueryRepository{messages: []port.MessageRecord{ + {UID: 1, ReceivedAt: time.Now(), Flags: []string{"\\Seen"}}, + {UID: 2, ReceivedAt: time.Now(), Flags: nil}, + }} + sess := newTestSession(auth, folders, messages, &fakeFlagRepository{}, &fakeBlobReader{}) + _ = sess.Login("user@example.test", testPassword) + _, _ = sess.Select("INBOX", &imap.SelectOptions{}) + + data, err := sess.Search(imapserver.NumKindUID, &imap.SearchCriteria{ + Not: []imap.SearchCriteria{{Flag: []imap.Flag{imap.FlagSeen}}}, + }, &imap.SearchOptions{}) + if err != nil { + t.Fatalf("Search: %v", err) + } + uidSet := data.All.(imap.UIDSet) + nums, _ := uidSet.Nums() + if len(nums) != 1 || nums[0] != 2 { + t.Errorf("matched UIDs = %v, want [2] (NOT SEEN excludes the \\Seen message)", nums) + } +} diff --git a/services/protocols/internal/presentation/imap/stubs.go b/services/protocols/internal/presentation/imap/stubs.go index 9db04e1..d637db6 100644 --- a/services/protocols/internal/presentation/imap/stubs.go +++ b/services/protocols/internal/presentation/imap/stubs.go @@ -36,12 +36,3 @@ func (s *session) Poll(_ *imapserver.UpdateWriter, _ bool) error { return nil } func (s *session) Idle(_ *imapserver.UpdateWriter, _ <-chan struct{}) error { return errNotYetImplemented } -func (s *session) Expunge(_ *imapserver.ExpungeWriter, _ *imap.UIDSet) error { - return errNotYetImplemented -} -func (s *session) Search(_ imapserver.NumKind, _ *imap.SearchCriteria, _ *imap.SearchOptions) (*imap.SearchData, error) { - return nil, errNotYetImplemented -} -func (s *session) Copy(_ imap.NumSet, _ string) (*imap.CopyData, error) { - return nil, errNotYetImplemented -}