fix(smtp): keep a copy of every submitted message in the sender's Sent folder
A message sent over SMTP submission was delivered but left nothing behind: the account's Sent folder stayed empty. Webmail files its own copy after sending, and a mail client cannot do it either, since IMAP APPEND is not implemented, so nothing ever filled Sent for SMTP clients. - ProcessOutboundEmailInput.FileSentCopy asks Submit to file the signed message in the sender's Sent folder, as read (not unread), sharing the blob the delivery jobs already reference; one copy per submission, not per recipient - the SMTP submission session sets it; the webmail leaves it off because it files its own copy without the Bcc line, and asking twice would duplicate it - filed only after the message is durably queued, so a submission refused by the send limit leaves nothing behind; a failure to file is logged, never returned, because the mail is already on its way and the client would send it again Tests: unit cases (asked/not asked, shared blob, failure swallowed, limit refusal, no repository) and an end-to-end test on the real submission server that fails with "Sent holds 0 messages" without the fix. Full go test ./... (25 packages) with TEST_DATABASE_URL, go vet and lang-lint are clean.
This commit is contained in:
1 parent
110a240cd9
commit
9c8b9c00fc
5 files changed
+323
No files matched your search
@@ -6,6 +6,7 @@ import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"log/slog"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
@@ -48,6 +49,13 @@ type ProcessOutboundEmailInput struct {
|
||||
// MaxRecipientsPerHour is the sender's policy, carried from the
|
||||
// authenticated session.
|
||||
MaxRecipientsPerHour int
|
||||
// FileSentCopy asks for a copy of the message in the sender's Sent folder.
|
||||
// Set for SMTP submission: a mail client cannot file it itself, since IMAP
|
||||
// APPEND is not implemented, so without this nothing a client sent ever
|
||||
// showed up in Sent. The webmail leaves it off because it files its own copy
|
||||
// (one that omits the Bcc line), and asking twice would show every sent
|
||||
// message twice.
|
||||
FileSentCopy bool
|
||||
}
|
||||
|
||||
// ProcessOutboundEmailUseCase implements the submission half of PLAN.md
|
||||
@@ -57,6 +65,14 @@ type ProcessOutboundEmailUseCase struct {
|
||||
outbound port.OutboundRepository
|
||||
blobs port.BlobStorage
|
||||
signer port.DkimSigner
|
||||
// sentCopies files the sender's copy; nil leaves FileSentCopy a no-op
|
||||
sentCopies port.InboundMessageRepository
|
||||
}
|
||||
|
||||
// SetSentCopies gives the use case somewhere to file the sender's copy of what
|
||||
// was submitted (see ProcessOutboundEmailInput.FileSentCopy).
|
||||
func (uc *ProcessOutboundEmailUseCase) SetSentCopies(messages port.InboundMessageRepository) {
|
||||
uc.sentCopies = messages
|
||||
}
|
||||
|
||||
func NewProcessOutboundEmailUseCase(
|
||||
@@ -143,9 +159,40 @@ func (uc *ProcessOutboundEmailUseCase) Submit(ctx context.Context, input Process
|
||||
}
|
||||
}
|
||||
|
||||
// Only once the message is durably queued, so a submission refused by the
|
||||
// send limit or a failed enqueue never leaves a "sent" copy behind. A
|
||||
// failure here is logged and swallowed: the mail is on its way, and telling
|
||||
// the client the submission failed would only make it send it again.
|
||||
if input.FileSentCopy && uc.sentCopies != nil {
|
||||
if err := uc.fileSentCopy(ctx, input, blob, payload); err != nil {
|
||||
slog.Warn("could not file the Sent copy", "mailbox", input.SenderAddr, "error", err)
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// fileSentCopy records the sender's copy in their Sent folder, referencing the
|
||||
// blob the delivery jobs already share, so it costs a row and no extra bytes.
|
||||
func (uc *ProcessOutboundEmailUseCase) fileSentCopy(ctx context.Context, input ProcessOutboundEmailInput, blob port.BlobRef, payload []byte) error {
|
||||
headers := ExtractMessageHeaders(payload)
|
||||
_, err := uc.sentCopies.PersistAll(ctx, []port.PersistInboundMessageInput{{
|
||||
MailboxID: input.MailboxID,
|
||||
Blob: blob,
|
||||
SenderAddress: input.SenderAddr,
|
||||
RecipientAddress: input.SenderAddr,
|
||||
TargetFolderName: "Sent",
|
||||
Subject: headers.Subject,
|
||||
Snippet: headers.Snippet,
|
||||
FromDisplayName: headers.FromDisplayName,
|
||||
MessageIDHeader: headers.MessageID,
|
||||
HasAttachments: headers.HasAttachments,
|
||||
// The sender's own copy is not unread mail
|
||||
AlreadySeen: true,
|
||||
}})
|
||||
return err
|
||||
}
|
||||
|
||||
// checkSendLimit refuses the whole submission when accepting it would take the
|
||||
// mailbox past its hourly recipient budget. Refusing up front is what contains
|
||||
// a compromised account; accepting and dropping later would not.
|
||||
|
||||
@@ -0,0 +1,139 @@
|
||||
package usecase
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"errors"
|
||||
"testing"
|
||||
|
||||
"github.com/google/uuid"
|
||||
|
||||
"lambdamail/protocols/internal/application/port"
|
||||
)
|
||||
|
||||
// recordingMessageRepo stands in for the repository that files a message into
|
||||
// a folder; it records what it was asked to store.
|
||||
type recordingMessageRepo struct {
|
||||
inputs []port.PersistInboundMessageInput
|
||||
err error
|
||||
}
|
||||
|
||||
func (r *recordingMessageRepo) PersistAll(_ context.Context, in []port.PersistInboundMessageInput) ([]int64, error) {
|
||||
if r.err != nil {
|
||||
return nil, r.err
|
||||
}
|
||||
r.inputs = append(r.inputs, in...)
|
||||
return make([]int64, len(in)), nil
|
||||
}
|
||||
|
||||
func sentCopyInput(mailboxID uuid.UUID, fileCopy bool) ProcessOutboundEmailInput {
|
||||
input := submissionInput(mailboxID, "ana@remote.test", "bruno@other.test")
|
||||
input.FileSentCopy = fileCopy
|
||||
input.Body = bytes.NewBufferString("From: user@example.test\r\nTo: ana@remote.test\r\nSubject: Aviso de teste\r\nMessage-ID: <abc@example.test>\r\n\r\nbody")
|
||||
return input
|
||||
}
|
||||
|
||||
// A mail client submitting over SMTP cannot file a copy in Sent itself (IMAP
|
||||
// APPEND is not implemented), so the server does it once the message is queued.
|
||||
func TestSubmit_FilesACopyInSentWhenAsked(t *testing.T) {
|
||||
mailboxID := uuid.New()
|
||||
messages := &recordingMessageRepo{}
|
||||
blobs := &capturingBlobStorage{}
|
||||
|
||||
uc := NewProcessOutboundEmailUseCase(nil, &countingOutboundRepo{}, blobs, nil)
|
||||
uc.SetSentCopies(messages)
|
||||
|
||||
if err := uc.Submit(context.Background(), sentCopyInput(mailboxID, true)); err != nil {
|
||||
t.Fatalf("Submit: %v", err)
|
||||
}
|
||||
|
||||
if len(messages.inputs) != 1 {
|
||||
t.Fatalf("filed %d copies, want exactly 1 (not one per recipient)", len(messages.inputs))
|
||||
}
|
||||
got := messages.inputs[0]
|
||||
if got.MailboxID != mailboxID {
|
||||
t.Errorf("copy filed for mailbox %v, want the sender %v", got.MailboxID, mailboxID)
|
||||
}
|
||||
if got.TargetFolderName != "Sent" {
|
||||
t.Errorf("folder = %q, want Sent", got.TargetFolderName)
|
||||
}
|
||||
if !got.AlreadySeen {
|
||||
t.Error("the sender's own copy must not count as unread mail")
|
||||
}
|
||||
if got.Subject != "Aviso de teste" || got.MessageIDHeader != "abc@example.test" {
|
||||
t.Errorf("headers not extracted: subject=%q message-id=%q", got.Subject, got.MessageIDHeader)
|
||||
}
|
||||
if got.RecipientAddress != "user@example.test" {
|
||||
t.Errorf("recipient address = %q, want the sender's own address", got.RecipientAddress)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSubmit_SentCopySharesTheStoredMessageInsteadOfDuplicatingIt(t *testing.T) {
|
||||
repo := &countingOutboundRepo{}
|
||||
messages := &recordingMessageRepo{}
|
||||
|
||||
uc := NewProcessOutboundEmailUseCase(nil, repo, &capturingBlobStorage{}, nil)
|
||||
uc.SetSentCopies(messages)
|
||||
if err := uc.Submit(context.Background(), sentCopyInput(uuid.New(), true)); err != nil {
|
||||
t.Fatalf("Submit: %v", err)
|
||||
}
|
||||
|
||||
if messages.inputs[0].Blob.ID != repo.jobs[0].BlobID {
|
||||
t.Error("the Sent copy must reference the same blob as the delivery jobs")
|
||||
}
|
||||
}
|
||||
|
||||
// The webmail files its own copy (without the Bcc line the envelope carries),
|
||||
// so asking it to file another one would put every sent message in Sent twice.
|
||||
func TestSubmit_DoesNotFileACopyUnlessAsked(t *testing.T) {
|
||||
messages := &recordingMessageRepo{}
|
||||
|
||||
uc := NewProcessOutboundEmailUseCase(nil, &countingOutboundRepo{}, &capturingBlobStorage{}, nil)
|
||||
uc.SetSentCopies(messages)
|
||||
if err := uc.Submit(context.Background(), sentCopyInput(uuid.New(), false)); err != nil {
|
||||
t.Fatalf("Submit: %v", err)
|
||||
}
|
||||
|
||||
if len(messages.inputs) != 0 {
|
||||
t.Fatalf("filed %d copies although none was requested", len(messages.inputs))
|
||||
}
|
||||
}
|
||||
|
||||
// The mail is already queued when the copy is filed; failing the submission
|
||||
// now would make the client send it again.
|
||||
func TestSubmit_ASentCopyFailureDoesNotFailTheSubmission(t *testing.T) {
|
||||
repo := &countingOutboundRepo{}
|
||||
messages := &recordingMessageRepo{err: errors.New("quota exceeded")}
|
||||
|
||||
uc := NewProcessOutboundEmailUseCase(nil, repo, &capturingBlobStorage{}, nil)
|
||||
uc.SetSentCopies(messages)
|
||||
|
||||
if err := uc.Submit(context.Background(), sentCopyInput(uuid.New(), true)); err != nil {
|
||||
t.Fatalf("Submit must succeed even when the Sent copy cannot be filed: %v", err)
|
||||
}
|
||||
if len(repo.jobs) != 2 {
|
||||
t.Fatalf("the message was not queued: %d jobs", len(repo.jobs))
|
||||
}
|
||||
}
|
||||
|
||||
func TestSubmit_ASubmissionRefusedByTheSendLimitLeavesNothingInSent(t *testing.T) {
|
||||
messages := &recordingMessageRepo{}
|
||||
|
||||
uc := NewProcessOutboundEmailUseCase(nil, &countingOutboundRepo{recentCount: 199}, &capturingBlobStorage{}, nil)
|
||||
uc.SetSentCopies(messages)
|
||||
|
||||
if err := uc.Submit(context.Background(), sentCopyInput(uuid.New(), true)); !errors.Is(err, ErrSendLimitExceeded) {
|
||||
t.Fatalf("error = %v, want ErrSendLimitExceeded", err)
|
||||
}
|
||||
if len(messages.inputs) != 0 {
|
||||
t.Fatal("a message that was refused must not appear in Sent")
|
||||
}
|
||||
}
|
||||
|
||||
func TestSubmit_WorksWithoutASentCopyRepository(t *testing.T) {
|
||||
uc := NewProcessOutboundEmailUseCase(nil, &countingOutboundRepo{}, &capturingBlobStorage{}, nil)
|
||||
|
||||
if err := uc.Submit(context.Background(), sentCopyInput(uuid.New(), true)); err != nil {
|
||||
t.Fatalf("Submit without a Sent repository: %v", err)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,133 @@
|
||||
package e2e
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/tls"
|
||||
"net"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/alexedwards/argon2id"
|
||||
embeddedpostgres "github.com/fergusstrange/embedded-postgres"
|
||||
"github.com/google/uuid"
|
||||
|
||||
appusecase "lambdamail/protocols/internal/application/usecase"
|
||||
"lambdamail/protocols/internal/infrastructure/diskstorage"
|
||||
"lambdamail/protocols/internal/infrastructure/dkim"
|
||||
"lambdamail/protocols/internal/infrastructure/postgres"
|
||||
smtppresentation "lambdamail/protocols/internal/presentation/smtp"
|
||||
|
||||
gosmtp "github.com/emersion/go-smtp"
|
||||
)
|
||||
|
||||
// TestSubmissionFilesTheSentCopy drives the real submission server. A client
|
||||
// that sends over SMTP used to leave nothing in the sender's Sent folder: the
|
||||
// mail was delivered, but the account showed no sent mail at all, and a client
|
||||
// cannot fill the folder itself because IMAP APPEND is not implemented.
|
||||
func TestSubmissionFilesTheSentCopy(t *testing.T) {
|
||||
if testing.Short() {
|
||||
t.Skip("e2e test skipped in -short mode")
|
||||
}
|
||||
ctx := context.Background()
|
||||
|
||||
pg := embeddedpostgres.NewDatabase(embeddedpostgres.DefaultConfig().
|
||||
Port(15450).
|
||||
RuntimePath(t.TempDir()).
|
||||
StartTimeout(120 * time.Second))
|
||||
if err := pg.Start(); err != nil {
|
||||
t.Fatalf("start embedded postgres: %v", err)
|
||||
}
|
||||
defer pg.Stop()
|
||||
|
||||
pool, err := postgres.NewPool(ctx, "postgres://postgres:postgres@localhost:15450/postgres?sslmode=disable")
|
||||
if err != nil {
|
||||
t.Fatalf("connect pool: %v", err)
|
||||
}
|
||||
defer pool.Close()
|
||||
applyMigrations(t, ctx, pool)
|
||||
|
||||
const (
|
||||
address = "plataforma@sentcopy.test"
|
||||
password = "correct horse battery staple"
|
||||
)
|
||||
hash, err := argon2id.CreateHash(password, argon2id.DefaultParams)
|
||||
if err != nil {
|
||||
t.Fatalf("hash: %v", err)
|
||||
}
|
||||
|
||||
domainID, mailboxID := uuid.New(), uuid.New()
|
||||
pool.Exec(ctx, `INSERT INTO domains (id, name, punycode_name) VALUES ($1, 'sentcopy.test', 'sentcopy.test')`, domainID)
|
||||
pool.Exec(ctx, `INSERT INTO mailboxes (id, domain_id, local_part, email_address, password_hash) VALUES ($1, $2, 'plataforma', $3, $4)`,
|
||||
mailboxID, domainID, address, hash)
|
||||
pool.Exec(ctx, `INSERT INTO folders (mailbox_id, name, special_use) VALUES ($1, 'INBOX', 'inbox'), ($1, 'Sent', 'sent')`, mailboxID)
|
||||
|
||||
blobs := diskstorage.NewLocalDiskBlobStorage(pool, t.TempDir())
|
||||
submissionUC := appusecase.NewProcessOutboundEmailUseCase(postgres.NewAuthRepository(pool), postgres.NewOutboundRepository(pool), blobs, dkim.NewSigner(nil))
|
||||
submissionUC.SetSentCopies(postgres.NewInboundMessageRepository(pool))
|
||||
|
||||
server := gosmtp.NewServer(smtppresentation.NewSubmissionBackend(submissionUC))
|
||||
server.Domain = "mail.sentcopy.test"
|
||||
server.AllowInsecureAuth = false
|
||||
server.TLSConfig = &tls.Config{Certificates: []tls.Certificate{selfSignedTLSCert(t)}}
|
||||
|
||||
ln, err := net.Listen("tcp", "127.0.0.1:0")
|
||||
if err != nil {
|
||||
t.Fatalf("listen: %v", err)
|
||||
}
|
||||
defer ln.Close()
|
||||
go server.Serve(ln)
|
||||
|
||||
submitMessage(t, ln.Addr().String(), address, password, "ana@remote.test")
|
||||
submitMessage(t, ln.Addr().String(), address, password, "bruno@other.test")
|
||||
|
||||
var sent int
|
||||
var subject string
|
||||
if err := pool.QueryRow(ctx, `
|
||||
SELECT COUNT(*), COALESCE(MAX(m.subject), '')
|
||||
FROM email_messages m JOIN folders f ON f.id = m.folder_id
|
||||
WHERE m.mailbox_id = $1 AND f.special_use = 'sent'`, mailboxID).Scan(&sent, &subject); err != nil {
|
||||
t.Fatalf("count Sent messages: %v", err)
|
||||
}
|
||||
|
||||
if sent != 2 {
|
||||
t.Fatalf("Sent holds %d messages after two submissions, want 2", sent)
|
||||
}
|
||||
if subject != "e2e" {
|
||||
t.Errorf("Sent copy subject = %q, want the submitted one", subject)
|
||||
}
|
||||
|
||||
var unread, total int
|
||||
if err := pool.QueryRow(ctx, `SELECT unread_count, total_count FROM folders WHERE mailbox_id = $1 AND special_use = 'sent'`, mailboxID).Scan(&unread, &total); err != nil {
|
||||
t.Fatalf("read Sent counters: %v", err)
|
||||
}
|
||||
if unread != 0 {
|
||||
t.Errorf("Sent advertises %d unread messages: the sender's own copies are not unread mail", unread)
|
||||
}
|
||||
if total != 2 {
|
||||
t.Errorf("Sent total_count = %d, want 2", total)
|
||||
}
|
||||
|
||||
var inbox int
|
||||
if err := pool.QueryRow(ctx, `
|
||||
SELECT COUNT(*) FROM email_messages m JOIN folders f ON f.id = m.folder_id
|
||||
WHERE m.mailbox_id = $1 AND f.special_use = 'inbox'`, mailboxID).Scan(&inbox); err != nil {
|
||||
t.Fatalf("count INBOX messages: %v", err)
|
||||
}
|
||||
if inbox != 0 {
|
||||
t.Errorf("INBOX holds %d messages, want 0: the copy belongs in Sent only", inbox)
|
||||
}
|
||||
|
||||
// The delivery jobs and the Sent copy share one stored message per submission
|
||||
var blobs2, copies int
|
||||
if err := pool.QueryRow(ctx, `SELECT COUNT(DISTINCT blob_id) FROM outbound_jobs WHERE mailbox_id = $1`, mailboxID).Scan(&blobs2); err != nil {
|
||||
t.Fatalf("count queued blobs: %v", err)
|
||||
}
|
||||
if err := pool.QueryRow(ctx, `
|
||||
SELECT COUNT(DISTINCT m.blob_id) FROM email_messages m
|
||||
JOIN outbound_jobs j ON j.blob_id = m.blob_id WHERE m.mailbox_id = $1`, mailboxID).Scan(&copies); err != nil {
|
||||
t.Fatalf("count shared blobs: %v", err)
|
||||
}
|
||||
if copies != blobs2 {
|
||||
t.Errorf("the Sent copies reference %d of the %d queued messages' blobs, want all", copies, blobs2)
|
||||
}
|
||||
}
|
||||
@@ -122,6 +122,8 @@ func (s *submissionSession) Data(r io.Reader) error {
|
||||
Recipients: s.recipients,
|
||||
Body: r,
|
||||
MaxRecipientsPerHour: s.authenticated.MaxRecipientsPerHour,
|
||||
// A client cannot file its own copy over IMAP, so the server keeps it
|
||||
FileSentCopy: true,
|
||||
})
|
||||
if err == nil {
|
||||
return nil
|
||||
|
||||
Reference in new issue
Block a user