From 67d0c8aa5a6d10948d5c621d059d3d6b70fb4892 Mon Sep 17 00:00:00 2001 From: Lucas Rayan Date: Thu, 22 Oct 2026 16:38:36 -0300 Subject: [PATCH] feat(protocols): conversations - thread_id and in_reply_to written at delivery, list and read APIs A reply now joins the conversation of the message it answers (Message-ID, In-Reply-To, References), falling back to the subject for replies with no headers. /mail/threads lists a folder as conversations and /mail/thread/{id} returns one. Replying from the webmail flags the answered message. Co-Authored-By: Claude Sonnet 5.5 --- .../0017_conversation_threading.down.sql | 3 + migrations/0017_conversation_threading.up.sql | 41 +++ .../cmd/lambdamail-protocols/main.go | 1 + .../application/port/conversations.go | 68 ++++ .../port/inbound_message_repository.go | 4 + .../application/usecase/message_headers.go | 37 ++- .../usecase/message_headers_test.go | 40 ++- .../usecase/process_inbound_email.go | 2 + .../usecase/process_outbound_email.go | 2 + .../application/usecase/webmail_mail.go | 51 +++ .../postgres/conversation_thread.go | 96 ++++++ .../postgres/conversation_thread_test.go | 308 +++++++++++++++++ .../infrastructure/postgres/conversations.go | 310 ++++++++++++++++++ .../postgres/inbound_message_repository.go | 12 +- .../infrastructure/postgres/mail_store.go | 2 +- .../internal/presentation/http/mail_api.go | 43 +++ .../internal/presentation/http/router.go | 14 + 17 files changed, 1025 insertions(+), 9 deletions(-) create mode 100644 migrations/0017_conversation_threading.down.sql create mode 100644 migrations/0017_conversation_threading.up.sql create mode 100644 services/protocols/internal/application/port/conversations.go create mode 100644 services/protocols/internal/infrastructure/postgres/conversation_thread.go create mode 100644 services/protocols/internal/infrastructure/postgres/conversation_thread_test.go create mode 100644 services/protocols/internal/infrastructure/postgres/conversations.go diff --git a/migrations/0017_conversation_threading.down.sql b/migrations/0017_conversation_threading.down.sql new file mode 100644 index 0000000..a8ee189 --- /dev/null +++ b/migrations/0017_conversation_threading.down.sql @@ -0,0 +1,3 @@ +DROP INDEX IF EXISTS idx_msg_mailbox_thread; +DROP FUNCTION IF EXISTS lm_has_reply_prefix(TEXT); +DROP FUNCTION IF EXISTS lm_subject_key(TEXT); diff --git a/migrations/0017_conversation_threading.up.sql b/migrations/0017_conversation_threading.up.sql new file mode 100644 index 0000000..001742a --- /dev/null +++ b/migrations/0017_conversation_threading.up.sql @@ -0,0 +1,41 @@ +-- Conversations. +-- +-- email_messages.thread_id has existed since the first migration and nothing +-- filled it, and in_reply_to was only ever copied, never written. A reply and +-- the message it answered were therefore unrelated rows, which is why the +-- webmail could neither show a conversation nor say whether it had been +-- answered. +-- +-- The subject key lives in the database so delivery, the backfill below and +-- anything later share one definition of "the same subject": the prefixes mail +-- clients add in several languages are not part of it. +CREATE FUNCTION lm_subject_key(subject TEXT) RETURNS TEXT AS $$ + SELECT btrim(regexp_replace( + lower(coalesce(subject, '')), + '^(\s*(re|fw|fwd|enc|res|rv|aw|wg|tr|antw)(\[\d+\])?\s*:)+', '')) +$$ LANGUAGE SQL IMMUTABLE; + +CREATE FUNCTION lm_has_reply_prefix(subject TEXT) RETURNS BOOLEAN AS $$ + SELECT lower(coalesce(subject, '')) ~ '^\s*(re|fw|fwd|enc|res|rv|aw|wg|tr|antw)(\[\d+\])?\s*:' +$$ LANGUAGE SQL IMMUTABLE; + +-- Mail that arrived before replies carried their headers cannot be linked by +-- Message-ID, so it is grouped by subject, and only where a reply is actually +-- in the mailbox: two unrelated messages that merely share a subject stay apart. +-- Everything else keeps thread_id NULL, which the webmail reads as "a +-- conversation of its own" (COALESCE(thread_id, id)). +WITH groups AS ( + SELECT DISTINCT mailbox_id, lm_subject_key(subject) AS k + FROM email_messages + WHERE lm_has_reply_prefix(subject) AND lm_subject_key(subject) <> '' +) +UPDATE email_messages m + SET thread_id = md5(m.mailbox_id::text || '|' || g.k)::uuid + FROM groups g + WHERE m.mailbox_id = g.mailbox_id + AND lm_subject_key(m.subject) = g.k + AND m.thread_id IS NULL; + +-- The listing groups by this and filters by mailbox and folder. +CREATE INDEX idx_msg_mailbox_thread ON email_messages (mailbox_id, (COALESCE(thread_id, id))) + WHERE expunged_at IS NULL; diff --git a/services/protocols/cmd/lambdamail-protocols/main.go b/services/protocols/cmd/lambdamail-protocols/main.go index 056fe25..2779c19 100644 --- a/services/protocols/cmd/lambdamail-protocols/main.go +++ b/services/protocols/cmd/lambdamail-protocols/main.go @@ -209,6 +209,7 @@ func run(cfg config) { if cfg.JwtSecret == "" { log.Printf("JWT_SECRET is not set: the webmail mail API stays disabled") } + webmailUC.SetConversations(postgres.NewConversations(pool)) router.SetMailAPI(webmailUC, cfg.JwtSecret) // Folders the mailbox owner keeps for themselves, which the filing rules // can then name as a destination. diff --git a/services/protocols/internal/application/port/conversations.go b/services/protocols/internal/application/port/conversations.go new file mode 100644 index 0000000..264d493 --- /dev/null +++ b/services/protocols/internal/application/port/conversations.go @@ -0,0 +1,68 @@ +package port + +import ( + "context" + "time" +) + +// WebmailParticipant is someone who wrote in a conversation. +type WebmailParticipant struct { + Name string `json:"name"` + Address string `json:"address"` + // Me marks the mailbox owner, so the list can say "me" instead of a name + // the reader knows perfectly well. + Me bool `json:"me"` +} + +// WebmailThread is one row of the conversation list: the newest message in the +// folder being looked at, with what the rest of the conversation adds. +type WebmailThread struct { + ThreadID string `json:"thread_id"` + UID uint32 `json:"uid"` + Subject string `json:"subject"` + SenderAddress string `json:"sender_address"` + FromName string `json:"from_display_name"` + Snippet string `json:"snippet"` + ReceivedAt time.Time `json:"received_at"` + HasAttachments bool `json:"has_attachments"` + MessageCount int `json:"message_count"` + UnseenCount int `json:"unseen_count"` + Seen bool `json:"seen"` + Flagged bool `json:"flagged"` + Answered bool `json:"answered"` + Participants []WebmailParticipant `json:"participants"` +} + +// WebmailThreadMessage is one message inside a conversation. +type WebmailThreadMessage struct { + UID uint32 `json:"uid"` + Folder string `json:"folder"` + Subject string `json:"subject"` + SenderAddress string `json:"sender_address"` + FromName string `json:"from_display_name"` + Snippet string `json:"snippet"` + ReceivedAt time.Time `json:"received_at"` + HasAttachments bool `json:"has_attachments"` + Seen bool `json:"seen"` + Flagged bool `json:"flagged"` + // Answered is true when the mailbox owner replied, whether the \Answered + // flag was set or the reply is simply there in the conversation - a reply + // written in another mail program leaves no flag, only the message. + Answered bool `json:"answered"` + // Own marks a message the mailbox owner wrote. + Own bool `json:"own"` +} + +// ConversationReader serves the conversation views. +type ConversationReader interface { + // ListThreads returns the conversations that have a message in the folder, + // newest first by that message. + ListThreads(ctx context.Context, mailboxID, folderName, search string, limit, offset int) ([]WebmailThread, error) + // ThreadMessages returns every message of one conversation, oldest first. + // folderName is the folder being viewed: a conversation's messages in Trash + // and Junk are left out unless that is the folder. + ThreadMessages(ctx context.Context, mailboxID, folderName, threadID string) ([]WebmailThreadMessage, error) + // MarkAnsweredByMessageID sets \Answered on the message a reply was written + // to, wherever in the mailbox it is filed. + MarkAnsweredByMessageID(ctx context.Context, mailboxID, messageID string) error +} diff --git a/services/protocols/internal/application/port/inbound_message_repository.go b/services/protocols/internal/application/port/inbound_message_repository.go index 27e219a..aecec27 100644 --- a/services/protocols/internal/application/port/inbound_message_repository.go +++ b/services/protocols/internal/application/port/inbound_message_repository.go @@ -22,6 +22,10 @@ type PersistInboundMessageInput struct { FromDisplayName string MessageIDHeader string HasAttachments bool + // InReplyTo and References place the message in its conversation. Both + // hold Message-IDs without angle brackets. + InReplyTo string + References []string // AlreadySeen marks a copy the user themselves composed - their own Sent // copy, or a draft. It is not unread mail, and counting it as such is why // the Sent folder carried an unread badge that nothing could clear: no diff --git a/services/protocols/internal/application/usecase/message_headers.go b/services/protocols/internal/application/usecase/message_headers.go index 65c685e..b4cbd93 100644 --- a/services/protocols/internal/application/usecase/message_headers.go +++ b/services/protocols/internal/application/usecase/message_headers.go @@ -19,10 +19,15 @@ type MessageHeaders struct { Subject string FromDisplayName string // FromAddress is the address in From, lowercased, or empty when it does not parse. - FromAddress string - MessageID string - Snippet string - HasAttachments bool + FromAddress string + MessageID string + Snippet string + HasAttachments bool + // InReplyTo and References are the Message-IDs this message answers, with + // the angle brackets removed so they compare equal to MessageID. They are + // what ties a reply to the conversation it belongs to. + InReplyTo string + References []string } // wordDecoder tolerates the charsets that turn up in real mail. An unknown @@ -57,6 +62,10 @@ func ExtractMessageHeaders(payload []byte) MessageHeaders { out.Subject = truncateUTF8(decodeHeader(msg.Header.Get("Subject")), 998) out.MessageID = truncateUTF8(strings.Trim(msg.Header.Get("Message-ID"), "<>"), 998) + if ids := messageIDsIn(msg.Header.Get("In-Reply-To")); len(ids) > 0 { + out.InReplyTo = ids[0] + } + out.References = messageIDsIn(msg.Header.Get("References")) if from := msg.Header.Get("From"); from != "" { if addr, err := mail.ParseAddress(from); err == nil { @@ -152,3 +161,23 @@ func truncateUTF8(s string, max int) string { } return cut } + +// maxReferences bounds how much of a long chain is kept. The ids nearest the +// end name the messages closest to this one, which is where its parent is. +const maxReferences = 20 + +// messageIDsIn lists the tokens of an In-Reply-To or References header, +// without their brackets, in the order written. +func messageIDsIn(header string) []string { + var ids []string + for _, field := range strings.Fields(header) { + id := strings.Trim(field, "<>,;") + if id != "" && len(id) <= 998 && strings.Contains(field, "<") { + ids = append(ids, id) + } + } + if len(ids) > maxReferences { + ids = ids[len(ids)-maxReferences:] + } + return ids +} diff --git a/services/protocols/internal/application/usecase/message_headers_test.go b/services/protocols/internal/application/usecase/message_headers_test.go index e7ff4c8..bebeb3f 100644 --- a/services/protocols/internal/application/usecase/message_headers_test.go +++ b/services/protocols/internal/application/usecase/message_headers_test.go @@ -1,6 +1,10 @@ package usecase -import "testing" +import ( + "fmt" + "strings" + "testing" +) func TestExtractMessageHeaders_ReadsSubjectSenderAndSnippet(t *testing.T) { msg := "From: Alice Example \r\n" + @@ -142,3 +146,37 @@ func TestExtractMessageHeaders_AttachmentDetection(t *testing.T) { }) } } + +func TestHeadersCarryWhatAMessageAnswers(t *testing.T) { + raw := "Message-ID: \r\nIn-Reply-To: \r\nReferences: \r\n \r\nSubject: Re: Plan\r\n\r\nbody" + h := ExtractMessageHeaders([]byte(raw)) + if h.InReplyTo != "b@x.test" { + t.Errorf("InReplyTo = %q, want b@x.test", h.InReplyTo) + } + if len(h.References) != 2 || h.References[0] != "a@x.test" || h.References[1] != "b@x.test" { + t.Errorf("References = %v, want [a@x.test b@x.test]", h.References) + } + if h.MessageID != "c@x.test" { + t.Errorf("MessageID = %q", h.MessageID) + } +} + +func TestAMessageThatAnswersNothingHasNoReferences(t *testing.T) { + h := ExtractMessageHeaders([]byte("Subject: Hi\r\n\r\nbody")) + if h.InReplyTo != "" || len(h.References) != 0 { + t.Errorf("a first message claims to answer %q %v", h.InReplyTo, h.References) + } +} + +// A chain can run to hundreds of ids. The ones nearest the end name the parent. +func TestAVeryLongChainKeepsTheNewestIDs(t *testing.T) { + var b strings.Builder + b.WriteString("References:") + for i := 0; i < 50; i++ { + fmt.Fprintf(&b, " ", i) + } + h := ExtractMessageHeaders([]byte(b.String() + "\r\nSubject: x\r\n\r\nbody")) + if len(h.References) != maxReferences || h.References[len(h.References)-1] != "m49@x.test" { + t.Errorf("kept %d ids ending %q", len(h.References), h.References[len(h.References)-1]) + } +} diff --git a/services/protocols/internal/application/usecase/process_inbound_email.go b/services/protocols/internal/application/usecase/process_inbound_email.go index a341112..b0953f3 100644 --- a/services/protocols/internal/application/usecase/process_inbound_email.go +++ b/services/protocols/internal/application/usecase/process_inbound_email.go @@ -347,6 +347,8 @@ func (uc *ProcessInboundEmailUseCase) Handle(ctx context.Context, input ProcessI FromDisplayName: headers.FromDisplayName, MessageIDHeader: headers.MessageID, HasAttachments: headers.HasAttachments, + InReplyTo: headers.InReplyTo, + References: headers.References, AlreadySeen: alreadySeen, SPFResult: authResult.SPF, DKIMResult: authResult.DKIM, diff --git a/services/protocols/internal/application/usecase/process_outbound_email.go b/services/protocols/internal/application/usecase/process_outbound_email.go index 05c6bf8..f5d1d8e 100644 --- a/services/protocols/internal/application/usecase/process_outbound_email.go +++ b/services/protocols/internal/application/usecase/process_outbound_email.go @@ -187,6 +187,8 @@ func (uc *ProcessOutboundEmailUseCase) fileSentCopy(ctx context.Context, input P FromDisplayName: headers.FromDisplayName, MessageIDHeader: headers.MessageID, HasAttachments: headers.HasAttachments, + InReplyTo: headers.InReplyTo, + References: headers.References, // The sender's own copy is not unread mail AlreadySeen: true, }}) diff --git a/services/protocols/internal/application/usecase/webmail_mail.go b/services/protocols/internal/application/usecase/webmail_mail.go index 81aa880..0a908cb 100644 --- a/services/protocols/internal/application/usecase/webmail_mail.go +++ b/services/protocols/internal/application/usecase/webmail_mail.go @@ -37,6 +37,9 @@ type WebmailUseCase struct { // the submission use case, not here.) blobStore port.BlobStorage messages port.InboundMessageRepository + // conversations serves the conversation views and records which message a + // reply answered. Optional; see SetConversations. + conversations port.ConversationReader } func NewWebmailUseCase( @@ -91,6 +94,42 @@ func (uc *WebmailUseCase) ListMessages( return uc.repo.ListMessages(ctx, id, folder, search, limit, offset) } +// SetConversations enables the conversation views. Optional: without it the +// thread routes answer "not configured" and the webmail falls back to the +// plain message list. +func (uc *WebmailUseCase) SetConversations(c port.ConversationReader) { + uc.conversations = c +} + +// ErrConversationsDisabled is returned when no conversation reader was wired. +var ErrConversationsDisabled = errors.New("conversations are not enabled") + +func (uc *WebmailUseCase) ListThreads( + ctx context.Context, address, folder, search string, limit, offset int, +) ([]port.WebmailThread, error) { + if uc.conversations == nil { + return nil, ErrConversationsDisabled + } + id, err := uc.mailboxID(ctx, address) + if err != nil { + return nil, err + } + return uc.conversations.ListThreads(ctx, id, folder, search, limit, offset) +} + +func (uc *WebmailUseCase) ThreadMessages( + ctx context.Context, address, folder, threadID string, +) ([]port.WebmailThreadMessage, error) { + if uc.conversations == nil { + return nil, ErrConversationsDisabled + } + id, err := uc.mailboxID(ctx, address) + if err != nil { + return nil, err + } + return uc.conversations.ThreadMessages(ctx, id, folder, threadID) +} + // GetMessage returns the raw RFC 5322 bytes and marks the message read. func (uc *WebmailUseCase) GetMessage(ctx context.Context, address, folder string, uid uint32) ([]byte, error) { id, err := uc.mailboxID(ctx, address) @@ -207,6 +246,16 @@ func (uc *WebmailUseCase) Send(ctx context.Context, input ComposeInput) error { // thing. Swallowed like the Sent copy above and for the same reason: the // mail is already queued, and failing the request now would only invite a // second copy to be sent. + // The message this one answers is marked answered. Swallowed for the same + // reason: the reply is already on its way, and a failed flag only means the + // list shows the conversation as unanswered until the next message. + if uc.conversations != nil && cleanMessageID(input.InReplyTo) != "" { + if mailboxID, idErr := uc.mailboxID(ctx, account.EmailAddress); idErr == nil { + if err := uc.conversations.MarkAnsweredByMessageID(ctx, mailboxID, input.InReplyTo); err != nil { + log.Printf("webmail: could not mark %s answered: %v", input.InReplyTo, err) + } + } + } if input.DraftUID != 0 { if mailboxID, idErr := uc.mailboxID(ctx, account.EmailAddress); idErr == nil { if err := uc.repo.Expunge(ctx, mailboxID, "Drafts", input.DraftUID); err != nil { @@ -341,6 +390,8 @@ func (uc *WebmailUseCase) SaveDraft(ctx context.Context, input ComposeInput, rep FromDisplayName: headers.FromDisplayName, MessageIDHeader: headers.MessageID, HasAttachments: headers.HasAttachments, + InReplyTo: headers.InReplyTo, + References: headers.References, AlreadySeen: true, }}) if err != nil { diff --git a/services/protocols/internal/infrastructure/postgres/conversation_thread.go b/services/protocols/internal/infrastructure/postgres/conversation_thread.go new file mode 100644 index 0000000..f13e893 --- /dev/null +++ b/services/protocols/internal/infrastructure/postgres/conversation_thread.go @@ -0,0 +1,96 @@ +package postgres + +import ( + "context" + "crypto/md5" + "errors" + "fmt" + "regexp" + "strings" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5" + + "lambdamail/protocols/internal/application/port" +) + +// resolveThread decides which conversation a message being filed belongs to. +// +// In order of how much each is to be trusted: +// +// 1. A message already in this mailbox with the same Message-ID. The same +// message filed twice - a copy in Sent and one delivered back to oneself, +// or a copy moved to another folder - is one conversation, not two. +// 2. The message it answers, found through In-Reply-To and then References. +// This is what actually ties a reply to its original. +// 3. A subject that starts with Re:, Fwd: or a translation of them, when the +// headers found nothing. The same key the migration used for older mail, +// so a reply to a message from before this existed still lands with it. +// 4. Otherwise it starts a conversation of its own. +// +// A message with no stored thread_id is read as a conversation of its own +// (COALESCE(thread_id, id)), so a parent from before threading is adopted by +// its own id and needs no rewrite. +func resolveThread(ctx context.Context, tx pgx.Tx, input port.PersistInboundMessageInput) (uuid.UUID, error) { + // The message's own id goes first so that it outranks the ones it answers. + var candidates []string + if input.MessageIDHeader != "" { + candidates = append(candidates, input.MessageIDHeader) + } + parents := make([]string, 0, len(input.References)+1) + if input.InReplyTo != "" { + parents = append(parents, input.InReplyTo) + } + // Nearest ancestor first: the last reference is the direct parent. + for i := len(input.References) - 1; i >= 0; i-- { + parents = append(parents, input.References[i]) + } + candidates = append(candidates, parents...) + + if len(candidates) > 0 { + var found uuid.UUID + err := tx.QueryRow(ctx, ` + SELECT COALESCE(thread_id, id) + FROM email_messages + WHERE mailbox_id = $1 + AND message_id_header = ANY($2) + AND expunged_at IS NULL + ORDER BY array_position($2::text[], message_id_header), received_at DESC + LIMIT 1`, input.MailboxID, candidates).Scan(&found) + if err == nil { + return found, nil + } + if !errors.Is(err, pgx.ErrNoRows) { + return uuid.Nil, fmt.Errorf("find the conversation: %w", err) + } + } + + if key := subjectKey(input.Subject); hasReplyPrefix(input.Subject) && key != "" { + return derivedThread(input.MailboxID, key), nil + } + return uuid.New(), nil +} + +// The subject helpers are written here as well as in the migration's SQL +// functions on purpose: delivery must keep working against a database whose +// migrations have not all run, so it cannot call functions a later migration +// creates. A test holds the two spellings to the same answers. +var replyPrefix = regexp.MustCompile(`^(\s*(re|fw|fwd|enc|res|rv|aw|wg|tr|antw)(\[\d+\])?\s*:)+`) + +func hasReplyPrefix(subject string) bool { + return replyPrefix.MatchString(strings.ToLower(subject)) +} + +// subjectKey is the subject with the prefixes mail clients add taken off. +func subjectKey(subject string) string { + return strings.TrimSpace(replyPrefix.ReplaceAllString(strings.ToLower(subject), "")) +} + +// derivedThread names the conversation a subject belongs to when nothing in +// the headers does. md5 over the same text as the migration's backfill, so +// mail filed before and after agree on it. +func derivedThread(mailboxID uuid.UUID, key string) uuid.UUID { + sum := md5.Sum([]byte(mailboxID.String() + "|" + key)) + id, _ := uuid.FromBytes(sum[:]) + return id +} diff --git a/services/protocols/internal/infrastructure/postgres/conversation_thread_test.go b/services/protocols/internal/infrastructure/postgres/conversation_thread_test.go new file mode 100644 index 0000000..f59821e --- /dev/null +++ b/services/protocols/internal/infrastructure/postgres/conversation_thread_test.go @@ -0,0 +1,308 @@ +package postgres + +import ( + "context" + "strings" + "testing" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgxpool" + + "lambdamail/protocols/internal/application/port" +) + +type threadFixture struct { + pool *pgxpool.Pool + repo *InboundMessageRepository + mailboxID uuid.UUID + domain string + blob port.BlobRef +} + +func newThreadFixture(t *testing.T) *threadFixture { + t.Helper() + pool := testPool(t) + t.Cleanup(pool.Close) + ctx := context.Background() + + domainID := uuid.New() + name := "thread-test-" + domainID.String() + ".invalid" + if _, err := pool.Exec(ctx, `INSERT INTO domains (id, name, punycode_name) VALUES ($1, $2, $3)`, domainID, name, name); err != nil { + t.Fatalf("seed domain: %v", err) + } + t.Cleanup(func() { pool.Exec(ctx, `DELETE FROM domains WHERE id = $1`, domainID) }) + + mailboxID := uuid.New() + if _, err := pool.Exec(ctx, `INSERT INTO mailboxes (id, domain_id, local_part, email_address, password_hash) VALUES ($1, $2, 'me', $3, 'x')`, + mailboxID, domainID, "me@"+name); err != nil { + t.Fatalf("seed mailbox: %v", err) + } + for _, f := range [][2]string{{"INBOX", "inbox"}, {"Sent", "sent"}} { + if _, err := pool.Exec(ctx, `INSERT INTO folders (id, mailbox_id, name, special_use) VALUES ($1, $2, $3, $4)`, + uuid.New(), mailboxID, f[0], f[1]); err != nil { + t.Fatalf("seed folder %s: %v", f[0], err) + } + } + blobID := uuid.New() + if _, 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, testBlobDigest()); err != nil { + t.Fatalf("seed blob: %v", err) + } + return &threadFixture{pool: pool, repo: NewInboundMessageRepository(pool), mailboxID: mailboxID, domain: name, + blob: port.BlobRef{ID: blobID, SHA256: "deadbeef", SizeBytes: 10}} +} + +// file stores one message and returns the conversation it landed in. +func (f *threadFixture) file(t *testing.T, in port.PersistInboundMessageInput) uuid.UUID { + t.Helper() + in.MailboxID = f.mailboxID + in.Blob = f.blob + if in.SenderAddress == "" { + in.SenderAddress = "someone@example.test" + } + in.RecipientAddress = "me@" + f.domain + in.SPFResult, in.DKIMResult, in.DMARCResult = "none", "none", "none" + if _, err := f.repo.PersistAll(context.Background(), []port.PersistInboundMessageInput{in}); err != nil { + t.Fatalf("PersistAll: %v", err) + } + var thread uuid.UUID + err := f.pool.QueryRow(context.Background(), ` + SELECT COALESCE(thread_id, id) FROM email_messages + WHERE mailbox_id = $1 AND subject = $2 ORDER BY received_at DESC, uid DESC LIMIT 1`, + f.mailboxID, in.Subject).Scan(&thread) + if err != nil { + t.Fatalf("read the conversation back: %v", err) + } + return thread +} + +func TestAReplyJoinsTheMessageItAnswers(t *testing.T) { + f := newThreadFixture(t) + root := f.file(t, port.PersistInboundMessageInput{Subject: "Lunch?", MessageIDHeader: "a@x.test"}) + reply := f.file(t, port.PersistInboundMessageInput{Subject: "Re: Lunch?", MessageIDHeader: "b@x.test", InReplyTo: "a@x.test"}) + if reply != root { + t.Fatalf("the reply started a conversation of its own") + } +} + +// The reply carries no Re: and a different subject, so only the headers can +// tie it to the conversation. +func TestHeadersOutrankTheSubject(t *testing.T) { + f := newThreadFixture(t) + root := f.file(t, port.PersistInboundMessageInput{Subject: "Lunch?", MessageIDHeader: "a@x.test"}) + reply := f.file(t, port.PersistInboundMessageInput{Subject: "Actually dinner", MessageIDHeader: "b@x.test", InReplyTo: "a@x.test"}) + if reply != root { + t.Fatalf("a reply with a changed subject left the conversation") + } +} + +// Only References survives some clients, and a long chain names the parent last. +func TestReferencesAloneFindTheConversation(t *testing.T) { + f := newThreadFixture(t) + root := f.file(t, port.PersistInboundMessageInput{Subject: "Plan", MessageIDHeader: "a@x.test"}) + f.file(t, port.PersistInboundMessageInput{Subject: "Re: Plan", MessageIDHeader: "b@x.test", InReplyTo: "a@x.test"}) + third := f.file(t, port.PersistInboundMessageInput{Subject: "Re: Re: Plan", MessageIDHeader: "c@x.test", + References: []string{"a@x.test", "b@x.test"}}) + if third != root { + t.Fatalf("a message known only by References left the conversation") + } +} + +// What was sent and what came back are one conversation. +func TestTheSentCopyJoinsTheConversation(t *testing.T) { + f := newThreadFixture(t) + root := f.file(t, port.PersistInboundMessageInput{Subject: "Quote", MessageIDHeader: "a@x.test"}) + sent := f.file(t, port.PersistInboundMessageInput{Subject: "Re: Quote", MessageIDHeader: "mine@x.test", + InReplyTo: "a@x.test", TargetFolderName: "Sent", AlreadySeen: true}) + if sent != root { + t.Fatalf("my own reply is not in the conversation it answers") + } +} + +func TestUnrelatedMessagesStayApart(t *testing.T) { + f := newThreadFixture(t) + one := f.file(t, port.PersistInboundMessageInput{Subject: "Invoice", MessageIDHeader: "a@x.test"}) + two := f.file(t, port.PersistInboundMessageInput{Subject: "Invoice", MessageIDHeader: "b@x.test"}) + if one == two { + t.Fatalf("two unrelated messages that share a subject were merged") + } +} + +// A reply that names nothing is still a reply by its subject. +func TestAReplyWithNoHeadersFallsBackToItsSubject(t *testing.T) { + f := newThreadFixture(t) + a := f.file(t, port.PersistInboundMessageInput{Subject: "Re: Status"}) + b := f.file(t, port.PersistInboundMessageInput{Subject: "RE: Re: STATUS"}) + if a != b { + t.Fatalf("replies to the same subject were not grouped") + } +} + +// The same message filed twice, such as a copy sent to oneself, is one conversation. +func TestTheSameMessageFiledTwiceIsOneConversation(t *testing.T) { + f := newThreadFixture(t) + a := f.file(t, port.PersistInboundMessageInput{Subject: "Note", MessageIDHeader: "n@x.test", TargetFolderName: "Sent", AlreadySeen: true}) + b := f.file(t, port.PersistInboundMessageInput{Subject: "Note", MessageIDHeader: "n@x.test"}) + if a != b { + t.Fatalf("the sent copy and the delivered copy are two conversations") + } +} + +// Mail from before threading has no thread_id and is its own conversation, so +// a reply must adopt it by id rather than orphan it. +func TestAReplyAdoptsAMessageFromBeforeThreading(t *testing.T) { + f := newThreadFixture(t) + f.file(t, port.PersistInboundMessageInput{Subject: "Old mail", MessageIDHeader: "old@x.test"}) + if _, err := f.pool.Exec(context.Background(), `UPDATE email_messages SET thread_id = NULL WHERE mailbox_id = $1`, f.mailboxID); err != nil { + t.Fatalf("simulate legacy: %v", err) + } + var legacyID uuid.UUID + if err := f.pool.QueryRow(context.Background(), `SELECT id FROM email_messages WHERE mailbox_id = $1`, f.mailboxID).Scan(&legacyID); err != nil { + t.Fatalf("read legacy id: %v", err) + } + reply := f.file(t, port.PersistInboundMessageInput{Subject: "Re: Old mail", MessageIDHeader: "new@x.test", InReplyTo: "old@x.test"}) + if reply != legacyID { + t.Fatalf("the reply did not adopt the older message") + } +} + +// --- reading conversations -------------------------------------------------- + +func (f *threadFixture) own(t *testing.T, in port.PersistInboundMessageInput) { + t.Helper() + in.TargetFolderName = "Sent" + in.AlreadySeen = true + in.SenderAddress = "me@" + f.domain + f.file(t, in) +} + +func (f *threadFixture) inbox(t *testing.T, from string, in port.PersistInboundMessageInput) { + t.Helper() + in.SenderAddress = from + f.file(t, in) +} + +func TestTheListHasOneRowPerConversationWithItsCount(t *testing.T) { + f := newThreadFixture(t) + f.inbox(t, "ana@x.test", port.PersistInboundMessageInput{Subject: "Plan", MessageIDHeader: "a@x.test"}) + f.own(t, port.PersistInboundMessageInput{Subject: "Re: Plan", MessageIDHeader: "b@x.test", InReplyTo: "a@x.test"}) + f.inbox(t, "ana@x.test", port.PersistInboundMessageInput{Subject: "Re: Plan", MessageIDHeader: "c@x.test", InReplyTo: "b@x.test"}) + f.inbox(t, "bob@x.test", port.PersistInboundMessageInput{Subject: "Other", MessageIDHeader: "z@x.test"}) + + threads, err := NewConversations(f.pool).ListThreads(context.Background(), f.mailboxID.String(), "INBOX", "", 50, 0) + if err != nil { + t.Fatalf("ListThreads: %v", err) + } + if len(threads) != 2 { + t.Fatalf("got %d rows, want 2 conversations", len(threads)) + } + plan := threads[0] + if !strings.Contains(plan.Subject, "Plan") { + plan = threads[1] + } + // The reply written from Sent counts towards the conversation shown in the inbox. + if plan.MessageCount != 3 { + t.Errorf("the conversation has %d messages, want 3 (two received, one sent)", plan.MessageCount) + } +} + +// What a reader wants to know at a glance. +func TestAConversationIsAnsweredWhenIReplied(t *testing.T) { + f := newThreadFixture(t) + f.inbox(t, "ana@x.test", port.PersistInboundMessageInput{Subject: "Quote", MessageIDHeader: "a@x.test"}) + conv := NewConversations(f.pool) + before, _ := conv.ListThreads(context.Background(), f.mailboxID.String(), "INBOX", "", 50, 0) + if before[0].Answered { + t.Fatalf("an unanswered conversation was shown as answered") + } + + // A reply written in another mail program leaves no \Answered flag, only the message. + f.own(t, port.PersistInboundMessageInput{Subject: "Re: Quote", MessageIDHeader: "b@x.test", InReplyTo: "a@x.test"}) + after, _ := conv.ListThreads(context.Background(), f.mailboxID.String(), "INBOX", "", 50, 0) + if !after[0].Answered { + t.Fatalf("the conversation I replied to is not shown as answered") + } +} + +func TestANewMessageAfterMyReplyIsUnansweredAgain(t *testing.T) { + f := newThreadFixture(t) + f.inbox(t, "ana@x.test", port.PersistInboundMessageInput{Subject: "Quote", MessageIDHeader: "a@x.test"}) + f.own(t, port.PersistInboundMessageInput{Subject: "Re: Quote", MessageIDHeader: "b@x.test", InReplyTo: "a@x.test"}) + f.inbox(t, "ana@x.test", port.PersistInboundMessageInput{Subject: "Re: Quote", MessageIDHeader: "c@x.test", InReplyTo: "b@x.test"}) + threads, _ := NewConversations(f.pool).ListThreads(context.Background(), f.mailboxID.String(), "INBOX", "", 50, 0) + if threads[0].Answered { + t.Fatalf("a conversation waiting on my reply was shown as answered") + } +} + +func TestMarkingAMessageAnsweredFlagsIt(t *testing.T) { + f := newThreadFixture(t) + f.inbox(t, "ana@x.test", port.PersistInboundMessageInput{Subject: "Ask", MessageIDHeader: "a@x.test"}) + conv := NewConversations(f.pool) + if err := conv.MarkAnsweredByMessageID(context.Background(), f.mailboxID.String(), ""); err != nil { + t.Fatalf("MarkAnsweredByMessageID: %v", err) + } + threads, _ := conv.ListThreads(context.Background(), f.mailboxID.String(), "INBOX", "", 50, 0) + msgs, err := conv.ThreadMessages(context.Background(), f.mailboxID.String(), "INBOX", threads[0].ThreadID) + if err != nil || len(msgs) != 1 || !msgs[0].Answered { + t.Fatalf("the flag did not show: %v %+v", err, msgs) + } +} + +func TestAConversationOfAnotherMailboxCannotBeRead(t *testing.T) { + f := newThreadFixture(t) + f.inbox(t, "ana@x.test", port.PersistInboundMessageInput{Subject: "Private", MessageIDHeader: "a@x.test"}) + conv := NewConversations(f.pool) + threads, _ := conv.ListThreads(context.Background(), f.mailboxID.String(), "INBOX", "", 50, 0) + + other := newThreadFixture(t) + if _, err := conv.ThreadMessages(context.Background(), other.mailboxID.String(), "INBOX", threads[0].ThreadID); err == nil { + t.Fatalf("another mailbox read this conversation") + } +} + +func TestThreadMessagesAreOldestFirstAndLeaveOutTrash(t *testing.T) { + f := newThreadFixture(t) + f.inbox(t, "ana@x.test", port.PersistInboundMessageInput{Subject: "Topic", MessageIDHeader: "a@x.test"}) + f.own(t, port.PersistInboundMessageInput{Subject: "Re: Topic", MessageIDHeader: "b@x.test", InReplyTo: "a@x.test"}) + conv := NewConversations(f.pool) + threads, _ := conv.ListThreads(context.Background(), f.mailboxID.String(), "INBOX", "", 50, 0) + msgs, err := conv.ThreadMessages(context.Background(), f.mailboxID.String(), "INBOX", threads[0].ThreadID) + if err != nil || len(msgs) != 2 { + t.Fatalf("got %d messages (%v), want 2", len(msgs), err) + } + if msgs[0].Own || !msgs[1].Own || !msgs[0].Answered { + t.Errorf("order or flags wrong: %+v", msgs) + } +} + +// Delivery computes the subject key itself, and the migration backfilled older +// mail with SQL. If the two ever disagreed, a reply would miss the conversation +// its older messages were put in. +func TestGoAndSQLAgreeOnTheSubjectKey(t *testing.T) { + pool := testPool(t) + defer pool.Close() + mailbox := uuid.New() + for _, subject := range []string{ + "Plan", "Re: Plan", "RE: Re[2]: Fwd: Plan ", " re :plan", "AW: Plan", "ENC: Plan", "Reunion", "Re:", "", + "Fw:Fw: invoice 12", "Resumo: vendas", "Antw: Re: Hallo", + } { + var sqlKey string + var sqlPrefix bool + var sqlThread uuid.UUID + if err := pool.QueryRow(context.Background(), + `SELECT lm_subject_key($1), lm_has_reply_prefix($1), md5($2::text || '|' || lm_subject_key($1))::uuid`, + subject, mailbox).Scan(&sqlKey, &sqlPrefix, &sqlThread); err != nil { + t.Fatalf("%q: %v", subject, err) + } + if got := subjectKey(subject); got != sqlKey { + t.Errorf("%q: Go key %q, SQL key %q", subject, got, sqlKey) + } + if got := hasReplyPrefix(subject); got != sqlPrefix { + t.Errorf("%q: Go prefix %v, SQL prefix %v", subject, got, sqlPrefix) + } + if got := derivedThread(mailbox, sqlKey); got != sqlThread { + t.Errorf("%q: Go thread %v, SQL thread %v", subject, got, sqlThread) + } + } +} diff --git a/services/protocols/internal/infrastructure/postgres/conversations.go b/services/protocols/internal/infrastructure/postgres/conversations.go new file mode 100644 index 0000000..6c02649 --- /dev/null +++ b/services/protocols/internal/infrastructure/postgres/conversations.go @@ -0,0 +1,310 @@ +package postgres + +import ( + "context" + "errors" + "fmt" + "sort" + "strings" + "time" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgxpool" + + "lambdamail/protocols/internal/application/port" +) + +// Conversations reads the mailbox as conversations rather than as single +// messages. Writing is MailStore's job, as for every other change. +type Conversations struct { + pool *pgxpool.Pool + store *MailStore +} + +func NewConversations(pool *pgxpool.Pool) *Conversations { + return &Conversations{pool: pool, store: NewMailStore(pool)} +} + +// scopeRow is one message that counts towards a conversation. +type scopeRow struct { + threadID uuid.UUID + uid uint32 + folder string + folderID string + subject string + sender string + name string + snippet string + receivedAt time.Time + hasAttach bool + seen bool + flagged bool + answeredFlag bool + own bool +} + +// scopeSQL selects the messages of the given conversations that belong in a +// view of one folder: that folder's own messages, and the rest of the +// conversation wherever it is filed, except what has been thrown away. Trash, +// Junk and Drafts are left out of a conversation seen from elsewhere - a +// reply still being written, or one deleted, is not part of the thread a reader +// is looking at - and are the whole story when that is the folder viewed. +const scopeSQL = ` + SELECT COALESCE(e.thread_id, e.id), e.uid, fo.name, fo.id::text, + COALESCE(e.subject, ''), e.sender_address::text, + COALESCE(e.from_display_name, ''), COALESCE(e.snippet, ''), + e.received_at, e.has_attachments, + EXISTS (SELECT 1 FROM message_flags f WHERE f.message_id = e.id AND f.received_at = e.received_at AND f.flag = '\Seen'), + EXISTS (SELECT 1 FROM message_flags f WHERE f.message_id = e.id AND f.received_at = e.received_at AND f.flag = '\Flagged'), + EXISTS (SELECT 1 FROM message_flags f WHERE f.message_id = e.id AND f.received_at = e.received_at AND f.flag = '\Answered'), + e.sender_address::text = mb.email_address::text + FROM email_messages e + JOIN folders fo ON fo.id = e.folder_id + JOIN mailboxes mb ON mb.id = e.mailbox_id + WHERE e.mailbox_id = $1 + AND e.expunged_at IS NULL + AND COALESCE(e.thread_id, e.id) = ANY($3::uuid[]) + AND (fo.id::text = $2 + OR (COALESCE(fo.special_use, '') NOT IN ('trash', 'junk', 'drafts') + AND lower(fo.name) NOT IN ('trash', 'junk', 'drafts'))) + ORDER BY e.received_at ASC, e.uid ASC` + +func (c *Conversations) scope(ctx context.Context, mailboxID, folderID string, threads []uuid.UUID) (map[uuid.UUID][]scopeRow, error) { + rows, err := c.pool.Query(ctx, scopeSQL, mailboxID, folderID, threads) + if err != nil { + return nil, fmt.Errorf("read the conversations: %w", err) + } + defer rows.Close() + out := make(map[uuid.UUID][]scopeRow, len(threads)) + for rows.Next() { + var r scopeRow + var uid int64 + if err := rows.Scan(&r.threadID, &uid, &r.folder, &r.folderID, &r.subject, &r.sender, &r.name, + &r.snippet, &r.receivedAt, &r.hasAttach, &r.seen, &r.flagged, &r.answeredFlag, &r.own); err != nil { + return nil, err + } + r.uid = uint32(uid) + out[r.threadID] = append(out[r.threadID], r) + } + return out, rows.Err() +} + +func (c *Conversations) ListThreads( + ctx context.Context, mailboxID, folderName, search string, limit, offset int, +) ([]port.WebmailThread, error) { + if limit <= 0 || limit > 200 { + limit = 50 + } + if offset < 0 { + offset = 0 + } + folderID, err := c.store.ResolveFolder(ctx, mailboxID, folderName) + if errors.Is(err, port.ErrFolderMissing) { + return []port.WebmailThread{}, nil + } + if err != nil { + return nil, err + } + + // The page is chosen first, by the newest message each conversation has in + // this folder, and only then are those conversations read in full. + rows, err := c.pool.Query(ctx, ` + SELECT COALESCE(thread_id, id) + FROM email_messages + WHERE folder_id = $1 AND expunged_at IS NULL + AND ($2 = '' OR subject ILIKE '%' || $2 || '%' + OR sender_address::text ILIKE '%' || $2 || '%' + OR snippet ILIKE '%' || $2 || '%') + GROUP BY COALESCE(thread_id, id) + ORDER BY max(received_at) DESC, max(uid) DESC + LIMIT $3 OFFSET $4`, folderID, search, limit, offset) + if err != nil { + return nil, fmt.Errorf("list conversations: %w", err) + } + var page []uuid.UUID + for rows.Next() { + var id uuid.UUID + if err := rows.Scan(&id); err != nil { + rows.Close() + return nil, err + } + page = append(page, id) + } + rows.Close() + if err := rows.Err(); err != nil { + return nil, err + } + if len(page) == 0 { + return []port.WebmailThread{}, nil + } + + byThread, err := c.scope(ctx, mailboxID, folderID, page) + if err != nil { + return nil, err + } + out := make([]port.WebmailThread, 0, len(page)) + for _, id := range page { + if t, ok := summarise(id, byThread[id], folderID); ok { + out = append(out, t) + } + } + return out, nil +} + +// summarise folds a conversation's messages into its list row. It reports +// false for a conversation with nothing in the folder viewed. +func summarise(id uuid.UUID, msgs []scopeRow, folderID string) (port.WebmailThread, bool) { + var latestHere *scopeRow + for i := range msgs { + if msgs[i].folderID == folderID { + latestHere = &msgs[i] // oldest first, so the last one wins + } + } + if latestHere == nil { + return port.WebmailThread{}, false + } + + t := port.WebmailThread{ + ThreadID: id.String(), UID: latestHere.uid, Subject: latestHere.subject, + SenderAddress: latestHere.sender, FromName: latestHere.name, Snippet: latestHere.snippet, + ReceivedAt: latestHere.receivedAt, MessageCount: len(msgs), + Answered: answered(msgs), + Participants: participants(msgs), + } + for _, m := range msgs { + if !m.seen { + t.UnseenCount++ + } + t.Flagged = t.Flagged || m.flagged + t.HasAttachments = t.HasAttachments || m.hasAttach + } + t.Seen = t.UnseenCount == 0 + return t, true +} + +// answered reports whether the last thing someone else wrote has been replied +// to - by the flag, or by a message of the owner's that came after it. +func answered(msgs []scopeRow) bool { + lastIncoming := -1 + for i, m := range msgs { + if !m.own { + lastIncoming = i + } + } + if lastIncoming < 0 { + return false + } + if msgs[lastIncoming].answeredFlag { + return true + } + for _, m := range msgs[lastIncoming+1:] { + if m.own { + return true + } + } + return false +} + +// participants lists who wrote, in the order they first did, at most four. +func participants(msgs []scopeRow) []port.WebmailParticipant { + seen := map[string]bool{} + var out []port.WebmailParticipant + for _, m := range msgs { + key := strings.ToLower(m.sender) + if seen[key] || len(out) == 4 { + continue + } + seen[key] = true + name := m.name + if name == "" { + name = m.sender + } + out = append(out, port.WebmailParticipant{Name: name, Address: m.sender, Me: m.own}) + } + return out +} + +func (c *Conversations) ThreadMessages( + ctx context.Context, mailboxID, folderName, threadID string, +) ([]port.WebmailThreadMessage, error) { + id, err := uuid.Parse(threadID) + if err != nil { + return nil, ErrMessageNotFound + } + folderID, err := c.store.ResolveFolder(ctx, mailboxID, folderName) + if errors.Is(err, port.ErrFolderMissing) { + return nil, ErrMessageNotFound + } + if err != nil { + return nil, err + } + byThread, err := c.scope(ctx, mailboxID, folderID, []uuid.UUID{id}) + if err != nil { + return nil, err + } + msgs := byThread[id] + if len(msgs) == 0 { + // Another mailbox's conversation and one that does not exist look alike. + return nil, ErrMessageNotFound + } + + out := make([]port.WebmailThreadMessage, 0, len(msgs)) + for i, m := range msgs { + out = append(out, port.WebmailThreadMessage{ + UID: m.uid, Folder: m.folder, Subject: m.subject, SenderAddress: m.sender, FromName: m.name, + Snippet: m.snippet, ReceivedAt: m.receivedAt, HasAttachments: m.hasAttach, + Seen: m.seen, Flagged: m.flagged, Own: m.own, + Answered: !m.own && (m.answeredFlag || ownAfter(msgs, i)), + }) + } + sort.SliceStable(out, func(a, b int) bool { return out[a].ReceivedAt.Before(out[b].ReceivedAt) }) + return out, nil +} + +func ownAfter(msgs []scopeRow, i int) bool { + for _, m := range msgs[i+1:] { + if m.own { + return true + } + } + return false +} + +// MarkAnsweredByMessageID sets \Answered on the message a reply was written to. +func (c *Conversations) MarkAnsweredByMessageID(ctx context.Context, mailboxID, messageID string) error { + messageID = strings.Trim(strings.TrimSpace(messageID), "<>") + if messageID == "" { + return nil + } + rows, err := c.pool.Query(ctx, ` + SELECT fo.id::text, e.uid + FROM email_messages e JOIN folders fo ON fo.id = e.folder_id + WHERE e.mailbox_id = $1 AND e.message_id_header = $2 AND e.expunged_at IS NULL + AND COALESCE(fo.special_use, '') NOT IN ('sent', 'drafts', 'trash')`, mailboxID, messageID) + if err != nil { + return fmt.Errorf("find the message that was answered: %w", err) + } + type target struct { + folder string + uid int64 + } + var targets []target + for rows.Next() { + var t target + if err := rows.Scan(&t.folder, &t.uid); err != nil { + rows.Close() + return err + } + targets = append(targets, t) + } + rows.Close() + for _, t := range targets { + if err := func() error { + _, e := c.store.SetFlags(ctx, t.folder, uint32(t.uid), port.FlagOpAdd, []string{`\Answered`}, 0) + return e + }(); err != nil { + return fmt.Errorf("flag the answered message: %w", err) + } + } + return nil +} diff --git a/services/protocols/internal/infrastructure/postgres/inbound_message_repository.go b/services/protocols/internal/infrastructure/postgres/inbound_message_repository.go index a37a1df..d7f440a 100644 --- a/services/protocols/internal/infrastructure/postgres/inbound_message_repository.go +++ b/services/protocols/internal/infrastructure/postgres/inbound_message_repository.go @@ -124,6 +124,11 @@ func persistOne(ctx context.Context, tx pgx.Tx, input port.PersistInboundMessage // id and received_at come back because message_flags is keyed on both: // email_messages is partitioned by received_at, so the flag rows cannot be // written without the same timestamp the row landed on. + threadID, err := resolveThread(ctx, tx, input) + if err != nil { + return 0, err + } + var messageID uuid.UUID var receivedAt time.Time err = tx.QueryRow(ctx, ` @@ -131,8 +136,8 @@ func persistOne(ctx context.Context, tx pgx.Tx, input port.PersistInboundMessage mailbox_id, folder_id, uid, modseq, blob_id, sender_address, recipient_addresses, size_bytes, subject, snippet, from_display_name, message_id_header, has_attachments, - spf_result, dkim_result, dmarc_result - ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16) + spf_result, dkim_result, dmarc_result, in_reply_to, thread_id + ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18) RETURNING id, received_at `, input.MailboxID, folderID, uid, modseq, input.Blob.ID, input.SenderAddress, []string{input.RecipientAddress}, input.Blob.SizeBytes, @@ -147,7 +152,8 @@ func persistOne(ctx context.Context, tx pgx.Tx, input port.PersistInboundMessage // Sent copy and drafts have no authentication results by definition - // and it failed the whole insert. Sent stayed empty because that write // is deliberately non-fatal, and saving a draft reported an error. - nullIfEmpty(input.SPFResult), nullIfEmpty(input.DKIMResult), nullIfEmpty(input.DMARCResult)). + nullIfEmpty(input.SPFResult), nullIfEmpty(input.DKIMResult), nullIfEmpty(input.DMARCResult), + nullIfEmpty(input.InReplyTo), threadID). Scan(&messageID, &receivedAt) if err != nil { return 0, fmt.Errorf("insert email_messages: %w", err) diff --git a/services/protocols/internal/infrastructure/postgres/mail_store.go b/services/protocols/internal/infrastructure/postgres/mail_store.go index 1a0fb30..24375e3 100644 --- a/services/protocols/internal/infrastructure/postgres/mail_store.go +++ b/services/protocols/internal/infrastructure/postgres/mail_store.go @@ -426,7 +426,7 @@ func (s *MailStore) copyOne(ctx context.Context, sourceFolderID string, sourceUI SELECT $2::uuid, $3::uuid, $4::bigint, $5::bigint, 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, + from_display_name, subject, snippet, message_id_header, in_reply_to, COALESCE(thread_id, id), has_attachments, arc_result, dane_status, tls_version, spam_score, spam_verdict, virus_name FROM email_messages WHERE id = $1 AND received_at = $6 RETURNING id, received_at`, diff --git a/services/protocols/internal/presentation/http/mail_api.go b/services/protocols/internal/presentation/http/mail_api.go index cf45825..7ab8635 100644 --- a/services/protocols/internal/presentation/http/mail_api.go +++ b/services/protocols/internal/presentation/http/mail_api.go @@ -102,6 +102,47 @@ func (m *mailAPI) handleMessages(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusOK, messages) } +// handleThreads lists the folder as conversations, one row each. +func (m *mailAPI) handleThreads(w http.ResponseWriter, r *http.Request) { + session, ok := m.authenticate(w, r) + if !ok { + return + } + query := r.URL.Query() + folder := query.Get("folder") + if folder == "" { + folder = "inbox" + } + limit, _ := strconv.Atoi(query.Get("limit")) + offset, _ := strconv.Atoi(query.Get("offset")) + + threads, err := m.useCase.ListThreads(r.Context(), session.Email, folder, query.Get("q"), limit, offset) + if err != nil { + m.fail(w, err) + return + } + writeJSON(w, http.StatusOK, threads) +} + +// handleThread returns the messages of one conversation, oldest first. +func (m *mailAPI) handleThread(w http.ResponseWriter, r *http.Request) { + session, ok := m.authenticate(w, r) + if !ok { + return + } + threadID := strings.TrimPrefix(r.URL.Path, "/api/v1/mail/thread/") + folder := r.URL.Query().Get("folder") + if folder == "" { + folder = "inbox" + } + messages, err := m.useCase.ThreadMessages(r.Context(), session.Email, folder, threadID) + if err != nil { + m.fail(w, err) + return + } + writeJSON(w, http.StatusOK, messages) +} + func (m *mailAPI) handleMessage(w http.ResponseWriter, r *http.Request) { session, ok := m.authenticate(w, r) if !ok { @@ -482,6 +523,8 @@ func (m *mailAPI) decodeCompose( // fail maps domain errors onto status codes without leaking internals. func (m *mailAPI) fail(w http.ResponseWriter, err error) { switch { + case errors.Is(err, usecase.ErrConversationsDisabled): + writeError(w, http.StatusNotImplemented, "CONVERSATIONS_DISABLED", "Conversations are not enabled") case errors.Is(err, postgres.ErrMessageNotFound), errors.Is(err, usecase.ErrNoSuchMailbox): writeError(w, http.StatusNotFound, "NOT_FOUND", "Message not found") case errors.Is(err, usecase.ErrNoRecipients): diff --git a/services/protocols/internal/presentation/http/router.go b/services/protocols/internal/presentation/http/router.go index 37354c2..34253b1 100644 --- a/services/protocols/internal/presentation/http/router.go +++ b/services/protocols/internal/presentation/http/router.go @@ -98,6 +98,8 @@ func (r *Router) registerRoutes() { r.mux.HandleFunc("/api/v1/mail/folders", r.handleMailFolders) r.mux.HandleFunc("/api/v1/mail/messages", r.handleMailMessages) r.mux.HandleFunc("/api/v1/mail/message/", r.handleMailMessage) + r.mux.HandleFunc("/api/v1/mail/threads", r.handleMailThreads) + r.mux.HandleFunc("/api/v1/mail/thread/", r.handleMailThread) r.mux.HandleFunc("/api/v1/mail/seen", r.handleMailSeen) r.mux.HandleFunc("/api/v1/mail/flagged", r.handleMailFlagged) r.mux.HandleFunc("/api/v1/mail/send", r.handleMailSend) @@ -287,6 +289,18 @@ func (r *Router) handleMailFolders(w http.ResponseWriter, req *http.Request) { } } +func (r *Router) handleMailThreads(w http.ResponseWriter, req *http.Request) { + if r.mailReady(w) { + r.mail.handleThreads(w, req) + } +} + +func (r *Router) handleMailThread(w http.ResponseWriter, req *http.Request) { + if r.mailReady(w) { + r.mail.handleThread(w, req) + } +} + func (r *Router) handleMailMessages(w http.ResponseWriter, req *http.Request) { if r.mailReady(w) { r.mail.handleMessages(w, req)