From 6b1e4dc00aead7014f48982164fdf91c874a7fbb Mon Sep 17 00:00:00 2001 From: Lucas Rayan Date: Fri, 24 Jul 2026 11:43:17 -0300 Subject: [PATCH] #6: implement IMAP IDLE (RFC 2177) and CONDSTORE (RFC 7162) --- .../cmd/lambdamail-protocols/main.go | 3 + .../application/port/flag_repository.go | 6 +- .../port/imap_folder_repository.go | 1 + .../port/message_query_repository.go | 1 + .../application/usecase/imap_session.go | 40 +++- .../application/usecase/imap_session_test.go | 26 +++ .../usecase/mailbox_tracker_manager.go | 72 ++++++ .../usecase/process_inbound_email.go | 20 +- .../e2e/imap_idle_condstore_e2e_test.go | 213 ++++++++++++++++++ .../postgres/copy_repository.go | 11 +- .../postgres/expunge_repository.go | 2 +- .../postgres/flag_repository.go | 64 ++++-- .../postgres/flag_repository_test.go | 27 ++- .../postgres/imap_folder_repository.go | 8 +- .../postgres/inbound_message_repository.go | 16 +- .../postgres/message_query_repository.go | 6 +- .../internal/presentation/imap/backend.go | 81 ++++++- .../presentation/imap/backend_test.go | 8 +- .../internal/presentation/imap/search.go | 16 +- .../internal/presentation/imap/stubs.go | 13 +- 20 files changed, 539 insertions(+), 95 deletions(-) create mode 100644 services/protocols/internal/application/usecase/mailbox_tracker_manager.go create mode 100644 services/protocols/internal/e2e/imap_idle_condstore_e2e_test.go diff --git a/services/protocols/cmd/lambdamail-protocols/main.go b/services/protocols/cmd/lambdamail-protocols/main.go index 92a36cf..820f6eb 100644 --- a/services/protocols/cmd/lambdamail-protocols/main.go +++ b/services/protocols/cmd/lambdamail-protocols/main.go @@ -87,6 +87,7 @@ func main() { expungeRepo := postgres.NewExpungeRepository(pool) copyRepo := postgres.NewCopyRepository(pool) imapUseCase := usecase.NewImapSessionUseCase(authRepo, imapFolders, messageQuery, flagRepo, blobReader, expungeRepo, copyRepo) + useCase.SetTrackerManager(imapUseCase.GetTrackerManager(), imapFolders) imapServer := imapserver.New(&imapserver.Options{ NewSession: func(c *imapserver.Conn) (imapserver.Session, *imapserver.GreetingData, error) { @@ -96,6 +97,8 @@ func main() { imap.CapIMAP4rev1: {}, imap.CapMove: {}, imap.CapUIDPlus: {}, + imap.CapIdle: {}, + imap.CapCondStore: {}, }, TLSConfig: &tls.Config{GetCertificate: certProvider.GetCertificate, MinVersion: tls.VersionTLS12}, InsecureAuth: false, diff --git a/services/protocols/internal/application/port/flag_repository.go b/services/protocols/internal/application/port/flag_repository.go index 9a84e1f..1167cd9 100644 --- a/services/protocols/internal/application/port/flag_repository.go +++ b/services/protocols/internal/application/port/flag_repository.go @@ -16,6 +16,8 @@ const ( type FlagRepository interface { // SetFlags applies op with flags to the message identified by // (folderID, uid). Add/Del insert/delete individual message_flags rows; - // Set replaces the message's full flag set in one transaction. - SetFlags(ctx context.Context, folderID string, uid uint32, op FlagOp, flags []string) error + // Set replaces the message's full flag set in one transaction. If + // unchangedSince > 0 and the message's current modseq exceeds it, the update + // is skipped and updated returns false. + SetFlags(ctx context.Context, folderID string, uid uint32, op FlagOp, flags []string, unchangedSince uint64) (updated bool, err error) } diff --git a/services/protocols/internal/application/port/imap_folder_repository.go b/services/protocols/internal/application/port/imap_folder_repository.go index 38813f1..0d4ed9f 100644 --- a/services/protocols/internal/application/port/imap_folder_repository.go +++ b/services/protocols/internal/application/port/imap_folder_repository.go @@ -9,6 +9,7 @@ type ImapFolderRecord struct { UIDNext uint32 UIDValidity uint32 NumMessages uint32 + HighestModSeq uint64 } // ImapFolderRepository resolves a mailbox's folders for LIST and SELECT. diff --git a/services/protocols/internal/application/port/message_query_repository.go b/services/protocols/internal/application/port/message_query_repository.go index 598fe2f..c8ab8ae 100644 --- a/services/protocols/internal/application/port/message_query_repository.go +++ b/services/protocols/internal/application/port/message_query_repository.go @@ -14,6 +14,7 @@ type MessageRecord struct { SizeBytes int64 ReceivedAt time.Time Flags []string + ModSeq uint64 } // MessageQueryRepository lists a folder's messages for FETCH. diff --git a/services/protocols/internal/application/usecase/imap_session.go b/services/protocols/internal/application/usecase/imap_session.go index 31506a3..1441e19 100644 --- a/services/protocols/internal/application/usecase/imap_session.go +++ b/services/protocols/internal/application/usecase/imap_session.go @@ -23,17 +23,37 @@ var ErrNoSuchFolder = errors.New("imap session: no such folder") // HandleImapCommandUseCase (this sub-project's scope: LOGIN, LIST, SELECT, // FETCH, STORE; SEARCH/EXPUNGE/IDLE/MOVE land in later sub-projects). type ImapSessionUseCase struct { - auth port.AuthRepository - folders port.ImapFolderRepository - messages port.MessageQueryRepository - flags port.FlagRepository - blobs port.BlobReader - expunger port.ExpungeRepository - copier port.CopyRepository + auth port.AuthRepository + folders port.ImapFolderRepository + messages port.MessageQueryRepository + flags port.FlagRepository + blobs port.BlobReader + expunger port.ExpungeRepository + copier port.CopyRepository + trackerManager *MailboxTrackerManager } 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} + return &ImapSessionUseCase{ + auth: auth, + folders: folders, + messages: messages, + flags: flags, + blobs: blobs, + expunger: expunger, + copier: copier, + trackerManager: NewMailboxTrackerManager(), + } +} + +func (uc *ImapSessionUseCase) SetTrackerManager(tm *MailboxTrackerManager) { + if tm != nil { + uc.trackerManager = tm + } +} + +func (uc *ImapSessionUseCase) GetTrackerManager() *MailboxTrackerManager { + return uc.trackerManager } func (uc *ImapSessionUseCase) Login(ctx context.Context, address, password string) (string, error) { @@ -77,8 +97,8 @@ func (uc *ImapSessionUseCase) ReadBlob(ctx context.Context, blobID uuid.UUID) ([ return uc.blobs.ReadByID(ctx, blobID) } -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) SetFlags(ctx context.Context, folderID string, uid uint32, op port.FlagOp, flags []string, unchangedSince uint64) (bool, error) { + return uc.flags.SetFlags(ctx, folderID, uid, op, flags, unchangedSince) } func (uc *ImapSessionUseCase) Expunge(ctx context.Context, folderID string, uids []uint32) error { diff --git a/services/protocols/internal/application/usecase/imap_session_test.go b/services/protocols/internal/application/usecase/imap_session_test.go index 2c4fc3f..3e10f78 100644 --- a/services/protocols/internal/application/usecase/imap_session_test.go +++ b/services/protocols/internal/application/usecase/imap_session_test.go @@ -149,6 +149,32 @@ func TestImapSessionUseCase_Expunge_DelegatesToRepository(t *testing.T) { } } +type fakeFlagRepository struct { + lastFolderID string + lastUID uint32 + lastOp port.FlagOp + lastFlags []string + lastUnchangedSince uint64 +} + +func (f *fakeFlagRepository) SetFlags(_ context.Context, folderID string, uid uint32, op port.FlagOp, flags []string, unchangedSince uint64) (bool, error) { + f.lastFolderID, f.lastUID, f.lastOp, f.lastFlags, f.lastUnchangedSince = folderID, uid, op, flags, unchangedSince + return true, nil +} + +func TestImapSessionUseCase_SetFlags_DelegatesToRepository(t *testing.T) { + flagRepo := &fakeFlagRepository{} + uc := NewImapSessionUseCase(nil, nil, nil, flagRepo, nil, nil, nil) + + updated, err := uc.SetFlags(context.Background(), "folder-1", 7, port.FlagOpAdd, []string{"\\Seen"}, 10) + if err != nil || !updated { + t.Fatalf("SetFlags error: %v, updated=%v", err, updated) + } + if flagRepo.lastFolderID != "folder-1" || flagRepo.lastUID != 7 || flagRepo.lastUnchangedSince != 10 { + t.Errorf("flagRepo state = %+v, want folder-1 uid 7 unchangedSince 10", flagRepo) + } +} + 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) diff --git a/services/protocols/internal/application/usecase/mailbox_tracker_manager.go b/services/protocols/internal/application/usecase/mailbox_tracker_manager.go new file mode 100644 index 0000000..cd9a5fb --- /dev/null +++ b/services/protocols/internal/application/usecase/mailbox_tracker_manager.go @@ -0,0 +1,72 @@ +package usecase + +import ( + "sync" + + "github.com/emersion/go-imap/v2" + "github.com/emersion/go-imap/v2/imapserver" +) + +// MailboxTrackerManager maintains imapserver.MailboxTracker instances per folder ID. +// This allows active IMAP sessions to receive real-time notifications when emails arrive, +// flags change, or messages are expunged (IDLE / RFC 2177). +type MailboxTrackerManager struct { + mu sync.RWMutex + trackers map[string]*imapserver.MailboxTracker +} + +func NewMailboxTrackerManager() *MailboxTrackerManager { + return &MailboxTrackerManager{ + trackers: make(map[string]*imapserver.MailboxTracker), + } +} + +// GetTracker retrieves or initializes the MailboxTracker for folderID. +func (m *MailboxTrackerManager) GetTracker(folderID string, initialNumMessages uint32) *imapserver.MailboxTracker { + m.mu.Lock() + defer m.mu.Unlock() + + tracker, ok := m.trackers[folderID] + if !ok { + tracker = imapserver.NewMailboxTracker(initialNumMessages) + m.trackers[folderID] = tracker + } + return tracker +} + +// NotifyNumMessages informs all sessions watching folderID of a new EXISTS count. +func (m *MailboxTrackerManager) NotifyNumMessages(folderID string, n uint32) { + m.mu.RLock() + tracker, ok := m.trackers[folderID] + m.mu.RUnlock() + + if ok { + tracker.QueueNumMessages(n) + } +} + +// NotifyMessageFlags informs sessions watching folderID of updated flags on a message. +func (m *MailboxTrackerManager) NotifyMessageFlags(folderID string, seqNum uint32, uid uint32, flags []string, source *imapserver.SessionTracker) { + m.mu.RLock() + tracker, ok := m.trackers[folderID] + m.mu.RUnlock() + + if ok { + imapFlags := make([]imap.Flag, len(flags)) + for i, f := range flags { + imapFlags[i] = imap.Flag(f) + } + tracker.QueueMessageFlags(seqNum, imap.UID(uid), imapFlags, source) + } +} + +// NotifyExpunge informs sessions watching folderID that seqNum was expunged. +func (m *MailboxTrackerManager) NotifyExpunge(folderID string, seqNum uint32) { + m.mu.RLock() + tracker, ok := m.trackers[folderID] + m.mu.RUnlock() + + if ok { + tracker.QueueExpunge(seqNum) + } +} diff --git a/services/protocols/internal/application/usecase/process_inbound_email.go b/services/protocols/internal/application/usecase/process_inbound_email.go index fe021aa..4ef68c7 100644 --- a/services/protocols/internal/application/usecase/process_inbound_email.go +++ b/services/protocols/internal/application/usecase/process_inbound_email.go @@ -36,15 +36,22 @@ type ProcessInboundEmailInput struct { // resolution + durable persistence; SPF/DKIM/DMARC/spam/virus checks land in // later sub-projects). type ProcessInboundEmailUseCase struct { - mailboxes port.MailboxRepository - blobs port.BlobStorage - messages port.InboundMessageRepository + mailboxes port.MailboxRepository + blobs port.BlobStorage + messages port.InboundMessageRepository + folders port.ImapFolderRepository + trackerManager *MailboxTrackerManager } func NewProcessInboundEmailUseCase(mailboxes port.MailboxRepository, blobs port.BlobStorage, messages port.InboundMessageRepository) *ProcessInboundEmailUseCase { return &ProcessInboundEmailUseCase{mailboxes: mailboxes, blobs: blobs, messages: messages} } +func (uc *ProcessInboundEmailUseCase) SetTrackerManager(tm *MailboxTrackerManager, folders port.ImapFolderRepository) { + uc.trackerManager = tm + uc.folders = folders +} + // ResolveRecipient is called from the SMTP session's RCPT TO handler. It // returns every mailbox address should be delivered to (more than one for // an alias that fans out), or ErrRecipientNotFound / ErrMailboxQuotaExceeded. @@ -88,6 +95,13 @@ func (uc *ProcessInboundEmailUseCase) Handle(ctx context.Context, input ProcessI if err != nil { return err } + + if uc.trackerManager != nil && uc.folders != nil { + folderRec, err := uc.folders.FindByName(ctx, recipient.ID.String(), "INBOX") + if err == nil && folderRec != nil { + uc.trackerManager.NotifyNumMessages(folderRec.ID, folderRec.NumMessages) + } + } } return nil } diff --git a/services/protocols/internal/e2e/imap_idle_condstore_e2e_test.go b/services/protocols/internal/e2e/imap_idle_condstore_e2e_test.go new file mode 100644 index 0000000..2a32f55 --- /dev/null +++ b/services/protocols/internal/e2e/imap_idle_condstore_e2e_test.go @@ -0,0 +1,213 @@ +package e2e + +import ( + "bytes" + "context" + "net" + "os" + "path/filepath" + "testing" + "time" + + "github.com/alexedwards/argon2id" + embeddedpostgres "github.com/fergusstrange/embedded-postgres" + "github.com/emersion/go-imap/v2" + "github.com/emersion/go-imap/v2/imapclient" + "github.com/emersion/go-imap/v2/imapserver" + + appusecase "lambdamail/protocols/internal/application/usecase" + "lambdamail/protocols/internal/application/port" + "lambdamail/protocols/internal/infrastructure/diskstorage" + "lambdamail/protocols/internal/infrastructure/postgres" + imappresentation "lambdamail/protocols/internal/presentation/imap" +) + +func TestImapIdleAndCondStoreEndToEnd(t *testing.T) { + if testing.Short() { + t.Skip("e2e test skipped in -short mode") + } + ctx := context.Background() + + runtimeDir := t.TempDir() + pg := embeddedpostgres.NewDatabase(embeddedpostgres.DefaultConfig(). + Port(54331). // distinct port for concurrent e2e test execution + RuntimePath(runtimeDir). + StartTimeout(120 * time.Second)) + if err := pg.Start(); err != nil { + t.Fatalf("start embedded postgres: %v", err) + } + defer pg.Stop() + + dbURL := "postgres://postgres:postgres@localhost:54331/postgres?sslmode=disable" + pool, err := postgres.NewPool(ctx, dbURL) + if err != nil { + t.Fatalf("connect: %v", err) + } + defer pool.Close() + + root := repoRoot(t) + sql, err := os.ReadFile(filepath.Join(root, "migrations", "0001_init_schema.up.sql")) + if err != nil { + t.Fatalf("read migration: %v", err) + } + if _, err := pool.Exec(ctx, string(sql)); err != nil { + t.Fatalf("apply migration: %v", err) + } + + // --- Seed mailbox -------------------------------------------------------- + password := "correct horse battery staple" + hash, err := argon2id.CreateHash(password, argon2id.DefaultParams) + if err != nil { + t.Fatalf("hash password: %v", err) + } + domainID := "00000000-0000-0000-0000-000000000001" + mailboxID := "00000000-0000-0000-0000-000000000002" + folderID := "00000000-0000-0000-0000-000000000003" + if _, err := pool.Exec(ctx, `INSERT INTO domains (id, name, punycode_name) VALUES ($1, 'example.test', 'example.test')`, domainID); err != nil { + t.Fatalf("seed domain: %v", err) + } + if _, err := pool.Exec(ctx, `INSERT INTO mailboxes (id, domain_id, local_part, email_address, password_hash) VALUES ($1, $2, 'user', 'user@example.test', $3)`, mailboxID, domainID, hash); err != nil { + t.Fatalf("seed mailbox: %v", err) + } + if _, err := pool.Exec(ctx, `INSERT INTO folders (id, mailbox_id, name, special_use) VALUES ($1, $2, 'INBOX', 'inbox')`, folderID, mailboxID); err != nil { + t.Fatalf("seed folder: %v", err) + } + + spool := t.TempDir() + mailboxRepo := postgres.NewMailboxRepository(pool) + inboundUC := appusecase.NewProcessInboundEmailUseCase( + mailboxRepo, + diskstorage.NewLocalDiskBlobStorage(pool, spool), + postgres.NewInboundMessageRepository(pool), + ) + + imapFolderRepo := postgres.NewImapFolderRepository(pool) + imapUC := appusecase.NewImapSessionUseCase( + postgres.NewAuthRepository(pool), + imapFolderRepo, + postgres.NewMessageQueryRepository(pool), + postgres.NewFlagRepository(pool), + diskstorage.NewLocalDiskBlobReader(pool), + postgres.NewExpungeRepository(pool), + postgres.NewCopyRepository(pool), + ) + inboundUC.SetTrackerManager(imapUC.GetTrackerManager(), imapFolderRepo) + + server := imapserver.New(&imapserver.Options{ + NewSession: func(c *imapserver.Conn) (imapserver.Session, *imapserver.GreetingData, error) { + return imappresentation.NewSession(c, imapUC) + }, + Caps: imap.CapSet{ + imap.CapIMAP4rev1: {}, + imap.CapMove: {}, + imap.CapUIDPlus: {}, + imap.CapIdle: {}, + imap.CapCondStore: {}, + }, + InsecureAuth: true, + }) + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + go server.Serve(ln) + defer server.Close() + + // Deliver 1 initial message + recipientRecord, err := mailboxRepo.FindActiveByAddress(ctx, "user@example.test") + if err != nil || recipientRecord == nil { + t.Fatalf("find recipient record: %v", err) + } + testMessage := "From: sender@remote.example\r\nTo: user@example.test\r\nSubject: condstore test\r\n\r\nInitial email.\r\n" + if err := inboundUC.Handle(ctx, appusecase.ProcessInboundEmailInput{ + Sender: "sender@remote.example", + Recipients: []port.MailboxRecord{*recipientRecord}, + RecipientAddresses: []string{"user@example.test"}, + Body: bytes.NewBufferString(testMessage), + }); err != nil { + t.Fatalf("seed message: %v", err) + } + + // Connect Client A with UnilateralDataHandler to receive live IDLE updates + mailboxUpdates := make(chan *imapclient.UnilateralDataMailbox, 10) + clientOptions := &imapclient.Options{ + UnilateralDataHandler: &imapclient.UnilateralDataHandler{ + Mailbox: func(data *imapclient.UnilateralDataMailbox) { + mailboxUpdates <- data + }, + }, + } + clientA, err := imapclient.DialInsecure(ln.Addr().String(), clientOptions) + if err != nil { + t.Fatalf("dial clientA: %v", err) + } + defer clientA.Close() + + if err := clientA.Login("user@example.test", password).Wait(); err != nil { + t.Fatalf("LOGIN clientA: %v", err) + } + + selectData, err := clientA.Select("INBOX", nil).Wait() + if err != nil { + t.Fatalf("SELECT INBOX: %v", err) + } + + if selectData.NumMessages != 1 { + t.Fatalf("expected NumMessages=1, got %d", selectData.NumMessages) + } + + // FETCH message 1 + fetchItems, err := clientA.Fetch(imap.SeqSetNum(1), &imap.FetchOptions{Flags: true, UID: true}).Collect() + if err != nil { + t.Fatalf("FETCH message 1: %v", err) + } + if len(fetchItems) != 1 { + t.Fatalf("expected 1 fetch item, got %d", len(fetchItems)) + } + + // STORE flag on message 1 + if err := clientA.Store(imap.SeqSetNum(1), &imap.StoreFlags{Op: imap.StoreFlagsAdd, Flags: []imap.Flag{imap.FlagFlagged}}, nil).Close(); err != nil { + t.Fatalf("STORE flag failed: %v", err) + } + + // SEARCH test by flag + searchRes, err := clientA.Search(&imap.SearchCriteria{Flag: []imap.Flag{imap.FlagFlagged}}, nil).Wait() + if err != nil { + t.Fatalf("SEARCH FLAGGED: %v", err) + } + if searchRes.Count != 1 { + t.Fatalf("SEARCH FLAGGED expected 1 result, got %d", searchRes.Count) + } + + // --- IDLE test ----------------------------------------------------------- + idleCmd, err := clientA.Idle() + if err != nil { + t.Fatalf("start IDLE: %v", err) + } + + // Concurrently deliver a 2nd message while clientA is IDLE + newMessage := "From: sender2@remote.example\r\nTo: user@example.test\r\nSubject: idle update\r\n\r\nRealtime email.\r\n" + go func() { + time.Sleep(100 * time.Millisecond) + _ = inboundUC.Handle(context.Background(), appusecase.ProcessInboundEmailInput{ + Sender: "sender2@remote.example", + Recipients: []port.MailboxRecord{*recipientRecord}, + RecipientAddresses: []string{"user@example.test"}, + Body: bytes.NewBufferString(newMessage), + }) + }() + + // Wait for clientA to receive mailbox update notification via UnilateralDataHandler + select { + case update := <-mailboxUpdates: + if update == nil || update.NumMessages == nil || *update.NumMessages != 2 { + t.Fatalf("expected update.NumMessages=2, got %+v", update) + } + case <-time.After(3 * time.Second): + t.Fatal("timed out waiting for IDLE notification") + } + + if err := idleCmd.Close(); err != nil { + t.Fatalf("close IDLE: %v", err) + } +} diff --git a/services/protocols/internal/infrastructure/postgres/copy_repository.go b/services/protocols/internal/infrastructure/postgres/copy_repository.go index 7e0bcb2..7c6482d 100644 --- a/services/protocols/internal/infrastructure/postgres/copy_repository.go +++ b/services/protocols/internal/infrastructure/postgres/copy_repository.go @@ -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 { diff --git a/services/protocols/internal/infrastructure/postgres/expunge_repository.go b/services/protocols/internal/infrastructure/postgres/expunge_repository.go index cb61fd2..865c207 100644 --- a/services/protocols/internal/infrastructure/postgres/expunge_repository.go +++ b/services/protocols/internal/infrastructure/postgres/expunge_repository.go @@ -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) diff --git a/services/protocols/internal/infrastructure/postgres/flag_repository.go b/services/protocols/internal/infrastructure/postgres/flag_repository.go index 0f64161..cedb2c1 100644 --- a/services/protocols/internal/infrastructure/postgres/flag_repository.go +++ b/services/protocols/internal/infrastructure/postgres/flag_repository.go @@ -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(¤tModSeq); 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 } diff --git a/services/protocols/internal/infrastructure/postgres/flag_repository_test.go b/services/protocols/internal/infrastructure/postgres/flag_repository_test.go index 88be698..80a1d21 100644 --- a/services/protocols/internal/infrastructure/postgres/flag_repository_test.go +++ b/services/protocols/internal/infrastructure/postgres/flag_repository_test.go @@ -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) + } } diff --git a/services/protocols/internal/infrastructure/postgres/imap_folder_repository.go b/services/protocols/internal/infrastructure/postgres/imap_folder_repository.go index 034cb03..ee8a6b9 100644 --- a/services/protocols/internal/infrastructure/postgres/imap_folder_repository.go +++ b/services/protocols/internal/infrastructure/postgres/imap_folder_repository.go @@ -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) diff --git a/services/protocols/internal/infrastructure/postgres/inbound_message_repository.go b/services/protocols/internal/infrastructure/postgres/inbound_message_repository.go index 007dfe4..9ed7761 100644 --- a/services/protocols/internal/infrastructure/postgres/inbound_message_repository.go +++ b/services/protocols/internal/infrastructure/postgres/inbound_message_repository.go @@ -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 { diff --git a/services/protocols/internal/infrastructure/postgres/message_query_repository.go b/services/protocols/internal/infrastructure/postgres/message_query_repository.go index 3ac2479..09a0c0c 100644 --- a/services/protocols/internal/infrastructure/postgres/message_query_repository.go +++ b/services/protocols/internal/infrastructure/postgres/message_query_repository.go @@ -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) diff --git a/services/protocols/internal/presentation/imap/backend.go b/services/protocols/internal/presentation/imap/backend.go index 68941f5..58d3569 100644 --- a/services/protocols/internal/presentation/imap/backend.go +++ b/services/protocols/internal/presentation/imap/backend.go @@ -19,6 +19,7 @@ type session struct { useCase *usecase.ImapSessionUseCase mailboxID string selectedFolder *port.ImapFolderRecord + tracker *imapserver.SessionTracker } // NewSession is the imapserver.Options.NewSession callback. @@ -37,7 +38,13 @@ func toIMAPError(err error) error { } } -func (s *session) Close() error { return nil } +func (s *session) Close() error { + if s.tracker != nil { + s.tracker.Close() + s.tracker = nil + } + return nil +} func (s *session) Login(username, password string) error { mailboxID, err := s.useCase.Login(context.Background(), username, password) @@ -53,15 +60,27 @@ func (s *session) Select(mailbox string, _ *imap.SelectOptions) (*imap.SelectDat if err != nil { return nil, toIMAPError(err) } + if s.tracker != nil { + s.tracker.Close() + s.tracker = nil + } s.selectedFolder = rec + if tm := s.useCase.GetTrackerManager(); tm != nil { + s.tracker = tm.GetTracker(rec.ID, rec.NumMessages).NewSession() + } return &imap.SelectData{ - NumMessages: rec.NumMessages, - UIDNext: imap.UID(rec.UIDNext), - UIDValidity: rec.UIDValidity, + NumMessages: rec.NumMessages, + UIDNext: imap.UID(rec.UIDNext), + UIDValidity: rec.UIDValidity, + HighestModSeq: rec.HighestModSeq, }, nil } func (s *session) Unselect() error { + if s.tracker != nil { + s.tracker.Close() + s.tracker = nil + } s.selectedFolder = nil return nil } @@ -82,9 +101,26 @@ func (s *session) Status(mailbox string, options *imap.StatusOptions) (*imap.Sta if options.UIDValidity { data.UIDValidity = rec.UIDValidity } + if options.HighestModSeq { + data.HighestModSeq = rec.HighestModSeq + } return data, nil } +func (s *session) Poll(w *imapserver.UpdateWriter, allowExpunge bool) error { + if s.tracker == nil { + return nil + } + return s.tracker.Poll(w, allowExpunge) +} + +func (s *session) Idle(w *imapserver.UpdateWriter, stop <-chan struct{}) error { + if s.tracker == nil { + return nil + } + return s.tracker.Idle(w, stop) +} + func (s *session) List(w *imapserver.ListWriter, ref string, patterns []string, options *imap.ListOptions) error { recs, err := s.useCase.ListFolders(context.Background(), s.mailboxID) if err != nil { @@ -153,6 +189,9 @@ func (s *session) Fetch(w *imapserver.FetchWriter, numSet imap.NumSet, options * if fetchErr != nil { return } + if options.ChangedSince > 0 && msg.ModSeq <= options.ChangedSince { + return + } respWriter := w.CreateMessage(seqNum) fetchErr = s.writeFetchResponse(respWriter, msg, options) }) @@ -273,6 +312,11 @@ func (s *session) Expunge(w *imapserver.ExpungeWriter, uids *imap.UIDSet) error if err != nil { return err } + if tm := s.useCase.GetTrackerManager(); tm != nil && s.selectedFolder != nil { + for _, seqNum := range seqNums { + tm.NotifyExpunge(s.selectedFolder.ID, seqNum) + } + } // 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 - @@ -317,6 +361,9 @@ func (s *session) Copy(numSet imap.NumSet, dest string) (*imap.CopyData, error) sourceUIDs.AddNum(imap.UID(c.SourceUID)) destUIDs.AddNum(imap.UID(c.DestUID)) } + if tm := s.useCase.GetTrackerManager(); tm != nil { + tm.NotifyNumMessages(destFolder.ID, destFolder.NumMessages+uint32(len(copied))) + } return &imap.CopyData{UIDValidity: destFolder.UIDValidity, SourceUIDs: sourceUIDs, DestUIDs: destUIDs}, nil } @@ -345,6 +392,11 @@ func (s *session) Move(w *imapserver.MoveWriter, numSet imap.NumSet, dest string if err != nil { return err } + if tm := s.useCase.GetTrackerManager(); tm != nil && s.selectedFolder != nil { + for _, seqNum := range seqNums { + tm.NotifyExpunge(s.selectedFolder.ID, seqNum) + } + } // w is nil in unit tests that only assert the use-case call, mirroring // Expunge's nil-writer accommodation above. if w == nil { @@ -376,23 +428,30 @@ func (s *session) Store(w *imapserver.FetchWriter, numSet imap.NumSet, flags *im flagNames[i] = string(f) } + var unchangedSince uint64 + if options != nil { + unchangedSince = options.UnchangedSince + } + messages, err := s.useCase.FetchMessages(context.Background(), s.selectedFolder.ID) if err != nil { return toIMAPError(err) } - // Resolve numSet against the folder's real message list (same pattern as - // Fetch), rather than pulling raw values via NumSet.Nums(): dynamic - // ranges like "1:*" cannot be resolved without a concrete candidate to - // 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, messages, func(_ uint32, msg port.MessageRecord) { + s.numSetIter(numSet, messages, func(seqNum uint32, msg port.MessageRecord) { if storeErr != nil { return } - if err := s.useCase.SetFlags(context.Background(), s.selectedFolder.ID, msg.UID, op, flagNames); err != nil { + updated, err := s.useCase.SetFlags(context.Background(), s.selectedFolder.ID, msg.UID, op, flagNames, unchangedSince) + if err != nil { storeErr = toIMAPError(err) + return + } + if updated { + if tm := s.useCase.GetTrackerManager(); tm != nil { + tm.NotifyMessageFlags(s.selectedFolder.ID, seqNum, msg.UID, flagNames, s.tracker) + } } }) if storeErr != nil { diff --git a/services/protocols/internal/presentation/imap/backend_test.go b/services/protocols/internal/presentation/imap/backend_test.go index 03cdd67..5e478ef 100644 --- a/services/protocols/internal/presentation/imap/backend_test.go +++ b/services/protocols/internal/presentation/imap/backend_test.go @@ -69,10 +69,10 @@ type fakeFlagCall struct { flags []string } -func (f *fakeFlagRepository) SetFlags(_ context.Context, folderID string, uid uint32, op port.FlagOp, flags []string) error { +func (f *fakeFlagRepository) SetFlags(_ context.Context, folderID string, uid uint32, op port.FlagOp, flags []string, _ uint64) (bool, error) { f.lastFolderID, f.lastUID, f.lastOp, f.lastFlags = folderID, uid, op, flags f.calls = append(f.calls, fakeFlagCall{folderID: folderID, uid: uid, op: op, flags: flags}) - return nil + return true, nil } type fakeBlobReader struct { @@ -451,7 +451,7 @@ func TestSession_Expunge_CallsUseCaseWithDeletedUIDsOnly(t *testing.T) { 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"}}} + folders := &fakeImapFolderRepository{byName: map[string]port.ImapFolderRecord{"INBOX": {ID: "folder-1", NumMessages: 3}}} messages := &fakeMessageQueryRepository{messages: []port.MessageRecord{ {UID: 1, Flags: []string{"\\Deleted"}}, {UID: 2, Flags: nil}, @@ -534,7 +534,7 @@ func TestSession_Move_ExpungesCopiedMessageEvenWithoutDeletedFlag(t *testing.T) "user@example.test": {ID: mailboxID, PasswordHash: testPasswordHash}, }} folders := &fakeImapFolderRepository{byName: map[string]port.ImapFolderRecord{ - "INBOX": {ID: "folder-1"}, + "INBOX": {ID: "folder-1", NumMessages: 1}, "Archive": {ID: "folder-2", UIDValidity: 42}, }} // UID 5 has no \Deleted flag - MOVE must still expunge it from the diff --git a/services/protocols/internal/presentation/imap/search.go b/services/protocols/internal/presentation/imap/search.go index 1f9a774..66501d6 100644 --- a/services/protocols/internal/presentation/imap/search.go +++ b/services/protocols/internal/presentation/imap/search.go @@ -46,7 +46,7 @@ func (s *session) Search(numKind imapserver.NumKind, criteria *imap.SearchCriter return s.useCase.ReadBlob(context.Background(), blobID) } - if !matchCriteria(seqNum, msg.UID, msg.ReceivedAt, msg.SizeBytes, flags, criteria, loadRaw) { + if !matchCriteria(seqNum, msg.UID, msg.ReceivedAt, msg.SizeBytes, msg.ModSeq, flags, criteria, loadRaw) { continue } @@ -81,7 +81,7 @@ func (s *session) Search(numKind imapserver.NumKind, criteria *imap.SearchCriter // 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 { +func matchCriteria(seqNum, uid uint32, receivedAt time.Time, sizeBytes int64, modSeq uint64, flags []imap.Flag, criteria *imap.SearchCriteria, loadRaw func() ([]byte, error)) bool { for _, seqSet := range criteria.SeqNum { if seqNum == 0 || !seqSet.Contains(seqNum) { return false @@ -96,6 +96,12 @@ func matchCriteria(seqNum, uid uint32, receivedAt time.Time, sizeBytes int64, fl return false } + if criteria.ModSeq != nil && criteria.ModSeq.ModSeq > 0 { + if modSeq < criteria.ModSeq.ModSeq { + return false + } + } + flagSet := make(map[imap.Flag]struct{}, len(flags)) for _, f := range flags { flagSet[canonicalFlag(f)] = struct{}{} @@ -159,13 +165,13 @@ func matchCriteria(seqNum, uid uint32, receivedAt time.Time, sizeBytes int64, fl } for _, not := range criteria.Not { - if matchCriteria(seqNum, uid, receivedAt, sizeBytes, flags, ¬, loadRaw) { + if matchCriteria(seqNum, uid, receivedAt, sizeBytes, modSeq, 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) { + if !matchCriteria(seqNum, uid, receivedAt, sizeBytes, modSeq, flags, &or[0], loadRaw) && + !matchCriteria(seqNum, uid, receivedAt, sizeBytes, modSeq, flags, &or[1], loadRaw) { return false } } diff --git a/services/protocols/internal/presentation/imap/stubs.go b/services/protocols/internal/presentation/imap/stubs.go index d637db6..059421e 100644 --- a/services/protocols/internal/presentation/imap/stubs.go +++ b/services/protocols/internal/presentation/imap/stubs.go @@ -9,7 +9,6 @@ package imappresentation import ( "github.com/emersion/go-imap/v2" - "github.com/emersion/go-imap/v2/imapserver" ) var errNotYetImplemented = &imap.Error{ @@ -25,14 +24,4 @@ func (s *session) Unsubscribe(_ string) error { return errN func (s *session) Append(_ string, _ imap.LiteralReader, _ *imap.AppendOptions) (*imap.AppendData, error) { return nil, errNotYetImplemented } -// Poll is called by imapserver before the status response of nearly every -// command in the Authenticated/Selected state (see conn.go's c.poll), not -// just in response to an explicit client IDLE/NOOP - returning an error here -// would fail every real command (FETCH, STORE, ...), not just polling. This -// sub-project doesn't push unsolicited EXISTS/EXPUNGE/FETCH updates (no -// IDLE support - see Idle below), so reporting "no updates" via a nil -// return is the correct, spec-compliant no-op, not a stub. -func (s *session) Poll(_ *imapserver.UpdateWriter, _ bool) error { return nil } -func (s *session) Idle(_ *imapserver.UpdateWriter, _ <-chan struct{}) error { - return errNotYetImplemented -} +