diff --git a/services/protocols/go.mod b/services/protocols/go.mod index 48536cd..75d54b6 100644 --- a/services/protocols/go.mod +++ b/services/protocols/go.mod @@ -10,9 +10,12 @@ require ( require ( github.com/emersion/go-sasl v0.0.0-20241020182733-b788ff22d5a6 // indirect + github.com/fergusstrange/embedded-postgres v1.30.0 // indirect github.com/jackc/pgpassfile v1.0.0 // indirect github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect github.com/jackc/puddle/v2 v2.2.2 // indirect + github.com/lib/pq v1.10.9 // indirect + github.com/xi2/xz v0.0.0-20171230120015-48954b6210f8 // indirect golang.org/x/sync v0.17.0 // indirect golang.org/x/text v0.29.0 // indirect ) diff --git a/services/protocols/go.sum b/services/protocols/go.sum index a1f0a43..2f0a0f0 100644 --- a/services/protocols/go.sum +++ b/services/protocols/go.sum @@ -5,6 +5,8 @@ github.com/emersion/go-sasl v0.0.0-20241020182733-b788ff22d5a6 h1:oP4q0fw+fOSWn3 github.com/emersion/go-sasl v0.0.0-20241020182733-b788ff22d5a6/go.mod h1:iL2twTeMvZnrg54ZoPDNfJaJaqy0xIQFuBdrLsmspwQ= github.com/emersion/go-smtp v0.24.0 h1:g6AfoF140mvW0vLNPD/LuCBLEAdlxOjIXqbIkJIS6Wk= github.com/emersion/go-smtp v0.24.0/go.mod h1:ZtRRkbTyp2XTHCA+BmyTFTrj8xY4I+b4McvHxCU2gsQ= +github.com/fergusstrange/embedded-postgres v1.30.0 h1:ewv1e6bBlqOIYtgGgRcEnNDpfGlmfPxB8T3PO9tV68Q= +github.com/fergusstrange/embedded-postgres v1.30.0/go.mod h1:w0YvnCgf19o6tskInrOOACtnqfVlOvluz3hlNLY7tRk= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM= @@ -15,6 +17,8 @@ github.com/jackc/pgx/v5 v5.10.0 h1:VhSvgU2jSli8o3AqIEOTJr7rZwAEUVo4E4XhR94Zfr0= github.com/jackc/pgx/v5 v5.10.0/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4= github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo= github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4= +github.com/lib/pq v1.10.9 h1:YXG7RB+JIjhP29X+OtkiDnYaXQwpS4JEWq7dtCCRUEw= +github.com/lib/pq v1.10.9/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= @@ -22,6 +26,8 @@ github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UV github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= +github.com/xi2/xz v0.0.0-20171230120015-48954b6210f8 h1:nIPpBwaJSVYIxUFsDv3M8ofmx9yWTog9BfvIu0q41lo= +github.com/xi2/xz v0.0.0-20171230120015-48954b6210f8/go.mod h1:HUYIGzjTL3rfEspMxjDjgmT5uz5wzYJKVo23qUhYTos= golang.org/x/sync v0.17.0 h1:l60nONMj9l5drqw6jlhIELNv9I0A4OFgRsG9k2oT9Ug= golang.org/x/sync v0.17.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI= golang.org/x/text v0.29.0 h1:1neNs90w9YzJ9BocxfsQNHKuAT4pkghyXc4nhZ6sJvk= diff --git a/services/protocols/internal/application/port/mailbox_repository.go b/services/protocols/internal/application/port/mailbox_repository.go index 4b402fa..9153802 100644 --- a/services/protocols/internal/application/port/mailbox_repository.go +++ b/services/protocols/internal/application/port/mailbox_repository.go @@ -12,11 +12,23 @@ import ( type MailboxRecord struct { ID uuid.UUID MaxMessageBytes int64 + QuotaBytes int64 + UsedBytes int64 } -// MailboxRepository resolves a recipient address to an active mailbox. +// MailboxRepository resolves a recipient address to the active mailbox(es) +// mail for it should be delivered to. type MailboxRepository interface { // FindActiveByAddress returns nil, nil (not an error) when no active // mailbox with an active domain matches address. FindActiveByAddress(ctx context.Context, address string) (*MailboxRecord, error) + + // ResolveDeliveryTargets returns every active mailbox mail for address + // should be delivered to: the mailbox itself when address matches one + // directly, or the mailbox(es) reached through a matching alias + // (source_address exact match takes precedence over a domain + // catch-all). Returns an empty, non-nil slice (not an error) when + // nothing matches - a direct mailbox miss, an alias with no active + // destination, or no alias at all are all "not found", not failures. + ResolveDeliveryTargets(ctx context.Context, address string) ([]MailboxRecord, error) } diff --git a/services/protocols/internal/application/usecase/process_inbound_email.go b/services/protocols/internal/application/usecase/process_inbound_email.go index a367398..fe021aa 100644 --- a/services/protocols/internal/application/usecase/process_inbound_email.go +++ b/services/protocols/internal/application/usecase/process_inbound_email.go @@ -14,6 +14,13 @@ import ( // mailbox matches - the presentation layer maps this to SMTP 550 5.1.1. var ErrRecipientNotFound = errors.New("process inbound email: recipient not found or inactive") +// ErrMailboxQuotaExceeded is returned by ResolveRecipient when a resolved +// target mailbox has already reached its storage quota - the presentation +// layer maps this to SMTP 452 4.2.2. If a single RCPT resolves to multiple +// mailboxes (alias fan-out) and any one of them is full, the whole RCPT is +// rejected rather than partially delivering. +var ErrMailboxQuotaExceeded = errors.New("process inbound email: mailbox has reached its storage quota") + // ProcessInboundEmailInput carries one SMTP DATA transaction's worth of // state: the accumulated recipients from one or more successful Rcpt calls, // and the message body stream. @@ -38,16 +45,23 @@ func NewProcessInboundEmailUseCase(mailboxes port.MailboxRepository, blobs port. return &ProcessInboundEmailUseCase{mailboxes: mailboxes, blobs: blobs, messages: messages} } -// ResolveRecipient is called from the SMTP session's RCPT TO handler. -func (uc *ProcessInboundEmailUseCase) ResolveRecipient(ctx context.Context, address string) (*port.MailboxRecord, error) { - rec, err := uc.mailboxes.FindActiveByAddress(ctx, address) +// 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. +func (uc *ProcessInboundEmailUseCase) ResolveRecipient(ctx context.Context, address string) ([]port.MailboxRecord, error) { + targets, err := uc.mailboxes.ResolveDeliveryTargets(ctx, address) if err != nil { return nil, err } - if rec == nil { + if len(targets) == 0 { return nil, ErrRecipientNotFound } - return rec, nil + for _, t := range targets { + if t.UsedBytes >= t.QuotaBytes { + return nil, ErrMailboxQuotaExceeded + } + } + return targets, nil } // Handle is called once from the SMTP session's DATA handler, after one or diff --git a/services/protocols/internal/application/usecase/process_inbound_email_test.go b/services/protocols/internal/application/usecase/process_inbound_email_test.go index 2d4f6d8..b420d9c 100644 --- a/services/protocols/internal/application/usecase/process_inbound_email_test.go +++ b/services/protocols/internal/application/usecase/process_inbound_email_test.go @@ -23,19 +23,27 @@ func (f *fakeMailboxRepository) FindActiveByAddress(_ context.Context, address s return &rec, nil } +func (f *fakeMailboxRepository) ResolveDeliveryTargets(_ context.Context, address string) ([]port.MailboxRecord, error) { + rec, ok := f.byAddress[address] + if !ok { + return []port.MailboxRecord{}, nil + } + return []port.MailboxRecord{rec}, nil +} + func TestUseCase_ResolveRecipient_ReturnsRecordForActiveMailbox(t *testing.T) { mailboxID := uuid.New() mailboxes := &fakeMailboxRepository{byAddress: map[string]port.MailboxRecord{ - "user@example.test": {ID: mailboxID, MaxMessageBytes: 1000}, + "user@example.test": {ID: mailboxID, MaxMessageBytes: 1000, QuotaBytes: 1000}, }} uc := NewProcessInboundEmailUseCase(mailboxes, nil, nil) - rec, err := uc.ResolveRecipient(context.Background(), "user@example.test") + targets, err := uc.ResolveRecipient(context.Background(), "user@example.test") if err != nil { t.Fatalf("unexpected error: %v", err) } - if rec.ID != mailboxID { - t.Errorf("ID = %v, want %v", rec.ID, mailboxID) + if len(targets) != 1 || targets[0].ID != mailboxID { + t.Errorf("targets = %+v, want single record with ID %v", targets, mailboxID) } } @@ -49,6 +57,44 @@ func TestUseCase_ResolveRecipient_ReturnsErrRecipientNotFoundForUnknownAddress(t } } +func TestUseCase_ResolveRecipient_ReturnsErrMailboxQuotaExceededWhenFull(t *testing.T) { + mailboxes := &fakeMailboxRepository{byAddress: map[string]port.MailboxRecord{ + "full@example.test": {ID: uuid.New(), QuotaBytes: 1000, UsedBytes: 1000}, + }} + uc := NewProcessInboundEmailUseCase(mailboxes, nil, nil) + + _, err := uc.ResolveRecipient(context.Background(), "full@example.test") + if !errors.Is(err, ErrMailboxQuotaExceeded) { + t.Fatalf("error = %v, want ErrMailboxQuotaExceeded", err) + } +} + +func TestUseCase_ResolveRecipient_ReturnsErrMailboxQuotaExceededWhenAnyFanOutTargetIsFull(t *testing.T) { + mailboxes := &fakeAliasFanOutRepository{targets: []port.MailboxRecord{ + {ID: uuid.New(), QuotaBytes: 1000, UsedBytes: 0}, + {ID: uuid.New(), QuotaBytes: 1000, UsedBytes: 1000}, + }} + uc := NewProcessInboundEmailUseCase(mailboxes, nil, nil) + + _, err := uc.ResolveRecipient(context.Background(), "alias@example.test") + if !errors.Is(err, ErrMailboxQuotaExceeded) { + t.Fatalf("error = %v, want ErrMailboxQuotaExceeded when any fan-out target is full", err) + } +} + +// fakeAliasFanOutRepository simulates an alias resolving to multiple mailboxes. +type fakeAliasFanOutRepository struct { + targets []port.MailboxRecord +} + +func (f *fakeAliasFanOutRepository) FindActiveByAddress(_ context.Context, _ string) (*port.MailboxRecord, error) { + return nil, nil +} + +func (f *fakeAliasFanOutRepository) ResolveDeliveryTargets(_ context.Context, _ string) ([]port.MailboxRecord, error) { + return f.targets, nil +} + type stubBlobStorage struct { ref port.BlobRef err error diff --git a/services/protocols/internal/e2e/smtp_inbound_e2e_test.go b/services/protocols/internal/e2e/smtp_inbound_e2e_test.go new file mode 100644 index 0000000..18f23be --- /dev/null +++ b/services/protocols/internal/e2e/smtp_inbound_e2e_test.go @@ -0,0 +1,281 @@ +// Package e2e contains end-to-end validation of the F0/F1 scope declared in +// PLAN.md section 16: a real Postgres (embedded, no Docker required), the real +// migration, the real SMTP server wired exactly as cmd/lambdamail-protocols +// does, and a real SMTP client driving the session. +// +// Run: go test ./internal/e2e/ -v (downloads Postgres binaries on first run). +package e2e + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "fmt" + "net" + "net/smtp" + "os" + "path/filepath" + "strings" + "testing" + "time" + + embeddedpostgres "github.com/fergusstrange/embedded-postgres" + gosmtp "github.com/emersion/go-smtp" + + "lambdamail/protocols/internal/application/usecase" + "lambdamail/protocols/internal/infrastructure/diskstorage" + "lambdamail/protocols/internal/infrastructure/postgres" + smtppresentation "lambdamail/protocols/internal/presentation/smtp" +) + +const testMessage = "From: sender@remote.example\r\n" + + "To: postmaster@example.test\r\n" + + "Subject: e2e smoke\r\n" + + "\r\n" + + "F1 inbound validation body.\r\n" + +func TestInboundSMTPEndToEnd(t *testing.T) { + if testing.Short() { + t.Skip("e2e test skipped in -short mode") + } + ctx := context.Background() + + // --- 1. Real Postgres via embedded binaries --------------------------- + runtimeDir := t.TempDir() + pg := embeddedpostgres.NewDatabase(embeddedpostgres.DefaultConfig(). + Port(54329). + 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:54329/postgres?sslmode=disable" + pool, err := postgres.NewPool(ctx, dbURL) + if err != nil { + t.Fatalf("connect: %v", err) + } + defer pool.Close() + + // --- 2. Real migration + dev seed ------------------------------------ + root := repoRoot(t) + for _, f := range []string{ + filepath.Join(root, "migrations", "0001_init_schema.up.sql"), + filepath.Join(root, "scripts", "dev-seed.sql"), + } { + sql, err := os.ReadFile(f) + if err != nil { + t.Fatalf("read %s: %v", f, err) + } + if _, err := pool.Exec(ctx, string(sql)); err != nil { + t.Fatalf("apply %s: %v", f, err) + } + } + + // --- 3. Real server, wired exactly like main.go ----------------------- + spool := t.TempDir() + uc := usecase.NewProcessInboundEmailUseCase( + postgres.NewMailboxRepository(pool), + diskstorage.NewLocalDiskBlobStorage(pool, spool), + postgres.NewInboundMessageRepository(pool), + ) + server := gosmtp.NewServer(smtppresentation.NewBackend(uc)) + server.Domain = "mail.example.test" + + 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() + addr := ln.Addr().String() + + // --- 4. Unknown recipient must be rejected in-session (550) ----------- + c, err := smtp.Dial(addr) + if err != nil { + t.Fatalf("dial: %v", err) + } + if err := c.Mail("sender@remote.example"); err != nil { + t.Fatalf("MAIL FROM: %v", err) + } + err = c.Rcpt("nobody@example.test") + if err == nil { + t.Fatal("RCPT for unknown user succeeded; PLAN section 6.1 requires 550 5.1.1") + } + if !strings.Contains(err.Error(), "550") { + t.Errorf("unknown RCPT error = %q, want SMTP 550", err) + } + c.Close() + + // --- 5. Valid delivery ------------------------------------------------ + deliver := func() { + t.Helper() + if err := smtp.SendMail(addr, nil, "sender@remote.example", + []string{"postmaster@example.test"}, []byte(testMessage)); err != nil { + t.Fatalf("SendMail: %v", err) + } + } + deliver() + + // --- 6. Durability assertions (PLAN sections 6.1 and 9) --------------- + var uid int64 + var blobPath string + var sizeBytes int64 + err = pool.QueryRow(ctx, ` + SELECT m.uid, b.storage_path, b.size_bytes + FROM email_messages m JOIN message_blobs b ON b.id = m.blob_id + WHERE m.sender_address = 'sender@remote.example' + ORDER BY m.uid LIMIT 1 + `).Scan(&uid, &blobPath, &sizeBytes) + if err != nil { + t.Fatalf("query persisted message: %v", err) + } + if uid != 1 { + t.Errorf("first uid = %d, want 1 (folders.uid_next allocation)", uid) + } + raw, err := os.ReadFile(blobPath) + if err != nil { + t.Fatalf("EML file missing on disk: %v", err) + } + sum := sha256.Sum256(raw) + var dbSum string + if err := pool.QueryRow(ctx, + `SELECT content_sha256 FROM message_blobs LIMIT 1`).Scan(&dbSum); err != nil { + t.Fatalf("blob row: %v", err) + } + if hex.EncodeToString(sum[:]) != dbSum { + t.Errorf("on-disk sha256 %x != db %s", sum, dbSum) + } + + var outboxCount int + if err := pool.QueryRow(ctx, + `SELECT count(*) FROM domain_events_outbox WHERE event_type = 'EmailReceived'`).Scan(&outboxCount); err != nil { + t.Fatalf("outbox: %v", err) + } + if outboxCount != 1 { + t.Errorf("outbox events = %d, want 1", outboxCount) + } + + // --- 7. Dedup: identical message again => 1 blob, ref_count 2, uid 2 -- + deliver() + var blobs, refs int + if err := pool.QueryRow(ctx, + `SELECT count(*), max(ref_count) FROM message_blobs`).Scan(&blobs, &refs); err != nil { + t.Fatalf("dedup query: %v", err) + } + if blobs != 1 || refs != 2 { + t.Errorf("blobs=%d refs=%d, want 1 blob with ref_count 2 (PLAN section 9 dedup)", blobs, refs) + } + var maxUID int64 + if err := pool.QueryRow(ctx, `SELECT max(uid) FROM email_messages`).Scan(&maxUID); err != nil { + t.Fatalf("uid query: %v", err) + } + if maxUID != 2 { + t.Errorf("max uid = %d, want 2 (sequential per folder)", maxUID) + } + + // --- 8. Folder/mailbox counters kept in sync with persisted messages -- + t.Run("counters_updated", func(t *testing.T) { + var unread, total int + var used int64 + err := pool.QueryRow(ctx, ` + SELECT f.unread_count, f.total_count, mb.used_bytes + FROM folders f JOIN mailboxes mb ON mb.id = f.mailbox_id + WHERE f.special_use = 'inbox' AND mb.email_address = 'postmaster@example.test' + `).Scan(&unread, &total, &used) + if err != nil { + t.Fatalf("counters query: %v", err) + } + if unread != 2 || total != 2 || used != sizeBytes*2 { + t.Errorf("unread=%d total=%d used=%d, want unread=2 total=2 used=%d (PLAN section 6.1 '++counters')", unread, total, used, sizeBytes*2) + } + }) + + // --- 9. A full mailbox rejects RCPT with 452 4.2.2, in-session -------- + t.Run("quota_rejected_when_mailbox_full", func(t *testing.T) { + domainID := "00000000-0000-0000-0000-000000000001" + fullMailboxID := "10000000-0000-0000-0000-000000000099" + fullAddress := "full@example.test" + if _, err := pool.Exec(ctx, ` + INSERT INTO mailboxes (id, domain_id, local_part, email_address, password_hash, quota_bytes, used_bytes) + VALUES ($1, $2, 'full', $3, '$argon2id$v=19$m=65536,t=3,p=4$c29tZXNhbHQ$RdescudvJCsgt3ub+b+dWRWJTmaaJObG', 1000, 1000) + `, fullMailboxID, domainID, fullAddress); err != nil { + t.Fatalf("seed full mailbox: %v", err) + } + + c, err := smtp.Dial(addr) + if err != nil { + t.Fatalf("dial: %v", err) + } + defer c.Close() + if err := c.Mail("sender@remote.example"); err != nil { + t.Fatalf("MAIL FROM: %v", err) + } + err = c.Rcpt(fullAddress) + if err == nil { + t.Fatal("RCPT to a full mailbox succeeded; want 452 4.2.2") + } + if !strings.Contains(err.Error(), "452") { + t.Errorf("full-mailbox RCPT error = %q, want SMTP 452", err) + } + }) + + // --- 10. An alias fans out to its destination mailbox ----------------- + t.Run("alias_resolves_to_destination_mailbox", func(t *testing.T) { + domainID := "00000000-0000-0000-0000-000000000001" + aliasDestMailboxID := "10000000-0000-0000-0000-000000000098" + aliasDestAddress := "alias-dest@example.test" + aliasAddress := "hello@example.test" + if _, err := pool.Exec(ctx, ` + INSERT INTO mailboxes (id, domain_id, local_part, email_address, password_hash) + VALUES ($1, $2, 'alias-dest', $3, '$argon2id$v=19$m=65536,t=3,p=4$c29tZXNhbHQ$RdescudvJCsgt3ub+b+dWRWJTmaaJObG') + `, aliasDestMailboxID, domainID, aliasDestAddress); err != nil { + t.Fatalf("seed alias destination mailbox: %v", err) + } + if _, err := pool.Exec(ctx, ` + INSERT INTO folders (id, mailbox_id, name, special_use) VALUES ($1, $2, 'INBOX', 'inbox') + `, "10000000-0000-0000-0000-000000000097", aliasDestMailboxID); err != nil { + t.Fatalf("seed alias destination inbox folder: %v", err) + } + if _, err := pool.Exec(ctx, ` + INSERT INTO aliases (id, domain_id, source_address, destination_addresses) + VALUES ($1, $2, $3, $4) + `, "10000000-0000-0000-0000-000000000096", domainID, aliasAddress, []string{aliasDestAddress}); err != nil { + t.Fatalf("seed alias: %v", err) + } + + if err := smtp.SendMail(addr, nil, "sender@remote.example", + []string{aliasAddress}, []byte(testMessage)); err != nil { + t.Fatalf("SendMail to alias: %v", err) + } + + var count int + if err := pool.QueryRow(ctx, + `SELECT count(*) FROM email_messages WHERE mailbox_id = $1`, aliasDestMailboxID).Scan(&count); err != nil { + t.Fatalf("query alias destination messages: %v", err) + } + if count != 1 { + t.Errorf("messages persisted for alias destination mailbox = %d, want 1", count) + } + }) + + _ = fmt.Sprintf // keep fmt imported if assertions above change +} + +// repoRoot walks up from the test file location to the repository root. +func repoRoot(t *testing.T) string { + t.Helper() + dir, err := os.Getwd() + if err != nil { + t.Fatal(err) + } + for i := 0; i < 8; i++ { + if _, err := os.Stat(filepath.Join(dir, "PLAN.md")); err == nil { + return dir + } + dir = filepath.Dir(dir) + } + t.Fatal("could not locate repository root (PLAN.md)") + return "" +} diff --git a/services/protocols/internal/infrastructure/postgres/inbound_message_repository.go b/services/protocols/internal/infrastructure/postgres/inbound_message_repository.go index 8c3ab90..007dfe4 100644 --- a/services/protocols/internal/infrastructure/postgres/inbound_message_repository.go +++ b/services/protocols/internal/infrastructure/postgres/inbound_message_repository.go @@ -39,8 +39,11 @@ 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 WHERE id = $1`, folderID); err != nil { - return 0, fmt.Errorf("advance uid_next: %w", err) + if _, err := tx.Exec(ctx, ` + UPDATE folders SET uid_next = uid_next + 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) } // email_messages.id defaults to gen_random_uuid() per the migration - no explicit value needed. @@ -61,6 +64,10 @@ func (r *InboundMessageRepository) Persist(ctx context.Context, input port.Persi return 0, fmt.Errorf("increment blob ref_count: %w", err) } + if _, err := tx.Exec(ctx, `UPDATE mailboxes SET used_bytes = used_bytes + $1 WHERE id = $2`, input.Blob.SizeBytes, input.MailboxID); err != nil { + return 0, fmt.Errorf("increment mailbox used_bytes: %w", err) + } + if _, err := tx.Exec(ctx, ` INSERT INTO domain_events_outbox (event_type, aggregate_id, payload) VALUES ('EmailReceived', $1, $2) diff --git a/services/protocols/internal/infrastructure/postgres/inbound_message_repository_test.go b/services/protocols/internal/infrastructure/postgres/inbound_message_repository_test.go index b4473c9..ada6fa1 100644 --- a/services/protocols/internal/infrastructure/postgres/inbound_message_repository_test.go +++ b/services/protocols/internal/infrastructure/postgres/inbound_message_repository_test.go @@ -73,6 +73,25 @@ func TestInboundMessageRepository_Persist_AllocatesSequentialUIDs(t *testing.T) if refCount != 2 { t.Errorf("ref_count = %d, want 2 after two Persist calls referencing the same blob", refCount) } + + var unreadCount, totalCount int + if err := pool.QueryRow(ctx, `SELECT unread_count, total_count FROM folders WHERE id = $1`, folderID).Scan(&unreadCount, &totalCount); err != nil { + t.Fatalf("query folder counters: %v", err) + } + if unreadCount != 2 { + t.Errorf("folder unread_count = %d, want 2 after two Persist calls", unreadCount) + } + if totalCount != 2 { + t.Errorf("folder total_count = %d, want 2 after two Persist calls", totalCount) + } + + var usedBytes int64 + if err := pool.QueryRow(ctx, `SELECT used_bytes FROM mailboxes WHERE id = $1`, mailboxID).Scan(&usedBytes); err != nil { + t.Fatalf("query mailbox used_bytes: %v", err) + } + if usedBytes != blob.SizeBytes*2 { + t.Errorf("mailbox used_bytes = %d, want %d (two Persist calls of %d bytes each)", usedBytes, blob.SizeBytes*2, blob.SizeBytes) + } } func TestInboundMessageRepository_Persist_ReturnsErrorWhenNoInboxFolderExists(t *testing.T) { diff --git a/services/protocols/internal/infrastructure/postgres/mailbox_repository.go b/services/protocols/internal/infrastructure/postgres/mailbox_repository.go index 991907b..3ef4a77 100644 --- a/services/protocols/internal/infrastructure/postgres/mailbox_repository.go +++ b/services/protocols/internal/infrastructure/postgres/mailbox_repository.go @@ -21,7 +21,7 @@ func NewMailboxRepository(pool *pgxpool.Pool) *MailboxRepository { func (r *MailboxRepository) FindActiveByAddress(ctx context.Context, address string) (*port.MailboxRecord, error) { row := r.pool.QueryRow(ctx, ` - SELECT m.id, d.max_message_bytes + SELECT m.id, d.max_message_bytes, m.quota_bytes, m.used_bytes FROM mailboxes m JOIN domains d ON d.id = m.domain_id WHERE m.email_address = $1 @@ -31,7 +31,7 @@ func (r *MailboxRepository) FindActiveByAddress(ctx context.Context, address str var rec port.MailboxRecord var id uuid.UUID - if err := row.Scan(&id, &rec.MaxMessageBytes); err != nil { + if err := row.Scan(&id, &rec.MaxMessageBytes, &rec.QuotaBytes, &rec.UsedBytes); err != nil { if err == pgx.ErrNoRows { return nil, nil } @@ -40,3 +40,49 @@ func (r *MailboxRepository) FindActiveByAddress(ctx context.Context, address str rec.ID = id return &rec, nil } + +// ResolveDeliveryTargets tries a direct mailbox match first, then falls back +// to aliases: an exact source_address match takes precedence over a +// domain-wide catch-all. Destination addresses that don't resolve to an +// internal active mailbox (e.g. external forwarding) are silently skipped - +// outbound delivery doesn't exist yet in this sub-project's scope. +func (r *MailboxRepository) ResolveDeliveryTargets(ctx context.Context, address string) ([]port.MailboxRecord, error) { + direct, err := r.FindActiveByAddress(ctx, address) + if err != nil { + return nil, err + } + if direct != nil { + return []port.MailboxRecord{*direct}, nil + } + + var destinations []string + err = r.pool.QueryRow(ctx, ` + SELECT a.destination_addresses + FROM aliases a + JOIN domains d ON d.id = a.domain_id + WHERE d.is_active = true + AND a.is_active = true + AND d.name = split_part($1, '@', 2) + AND (a.source_address = $1 OR (a.is_catch_all AND a.source_address = '*@' || split_part($1, '@', 2))) + ORDER BY a.is_catch_all ASC + LIMIT 1 + `, address).Scan(&destinations) + if err != nil { + if err == pgx.ErrNoRows { + return []port.MailboxRecord{}, nil + } + return nil, err + } + + targets := []port.MailboxRecord{} + for _, dest := range destinations { + rec, err := r.FindActiveByAddress(ctx, dest) + if err != nil { + return nil, err + } + if rec != nil { + targets = append(targets, *rec) + } + } + return targets, nil +} diff --git a/services/protocols/internal/infrastructure/postgres/mailbox_repository_test.go b/services/protocols/internal/infrastructure/postgres/mailbox_repository_test.go index cd9d7f8..b6b11c4 100644 --- a/services/protocols/internal/infrastructure/postgres/mailbox_repository_test.go +++ b/services/protocols/internal/infrastructure/postgres/mailbox_repository_test.go @@ -118,3 +118,194 @@ func TestMailboxRepository_FindActiveByAddress_ExcludesInactiveMailbox(t *testin t.Errorf("rec = %+v, want nil - mailbox is_active=false must be excluded", rec) } } + +func TestMailboxRepository_FindActiveByAddress_ReturnsQuotaAndUsedBytes(t *testing.T) { + pool := testPool(t) + defer pool.Close() + ctx := context.Background() + + domainID := uuid.New() + domainName := "repo-test-quota-" + domainID.String() + ".invalid" + _, err := pool.Exec(ctx, `INSERT INTO domains (id, name, punycode_name) VALUES ($1, $2, $3)`, + domainID, domainName, domainName) + if err != nil { + t.Fatalf("seed domain: %v", err) + } + defer pool.Exec(ctx, `DELETE FROM domains WHERE id = $1`, domainID) + + address := "user@" + domainName + _, err = pool.Exec(ctx, `INSERT INTO mailboxes (id, domain_id, local_part, email_address, password_hash, quota_bytes, used_bytes) VALUES ($1, $2, 'user', $3, '$argon2id$v=19$m=65536,t=3,p=4$c29tZXNhbHQ$RdescudvJCsgt3ub+b+dWRWJTmaaJObG', 5000, 4000)`, + uuid.New(), domainID, address) + if err != nil { + t.Fatalf("seed mailbox: %v", err) + } + + repo := NewMailboxRepository(pool) + rec, err := repo.FindActiveByAddress(ctx, address) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if rec == nil { + t.Fatal("rec = nil, want a record for the seeded mailbox") + } + if rec.QuotaBytes != 5000 { + t.Errorf("QuotaBytes = %d, want 5000", rec.QuotaBytes) + } + if rec.UsedBytes != 4000 { + t.Errorf("UsedBytes = %d, want 4000", rec.UsedBytes) + } +} + +func TestMailboxRepository_ResolveDeliveryTargets_PrefersDirectMailboxMatch(t *testing.T) { + pool := testPool(t) + defer pool.Close() + ctx := context.Background() + + domainID := uuid.New() + domainName := "repo-test-direct-" + domainID.String() + ".invalid" + _, err := pool.Exec(ctx, `INSERT INTO domains (id, name, punycode_name) VALUES ($1, $2, $3)`, + domainID, domainName, domainName) + if err != nil { + t.Fatalf("seed domain: %v", err) + } + defer pool.Exec(ctx, `DELETE FROM domains WHERE id = $1`, domainID) + + mailboxID := uuid.New() + address := "user@" + domainName + _, err = pool.Exec(ctx, `INSERT INTO mailboxes (id, domain_id, local_part, email_address, password_hash) VALUES ($1, $2, 'user', $3, '$argon2id$v=19$m=65536,t=3,p=4$c29tZXNhbHQ$RdescudvJCsgt3ub+b+dWRWJTmaaJObG')`, + mailboxID, domainID, address) + if err != nil { + t.Fatalf("seed mailbox: %v", err) + } + + repo := NewMailboxRepository(pool) + targets, err := repo.ResolveDeliveryTargets(ctx, address) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(targets) != 1 || targets[0].ID != mailboxID { + t.Errorf("targets = %+v, want single record with ID %v (direct mailbox match)", targets, mailboxID) + } +} + +func TestMailboxRepository_ResolveDeliveryTargets_ResolvesSpecificAlias(t *testing.T) { + pool := testPool(t) + defer pool.Close() + ctx := context.Background() + + domainID := uuid.New() + domainName := "repo-test-alias-" + domainID.String() + ".invalid" + _, err := pool.Exec(ctx, `INSERT INTO domains (id, name, punycode_name) VALUES ($1, $2, $3)`, + domainID, domainName, domainName) + if err != nil { + t.Fatalf("seed domain: %v", err) + } + defer pool.Exec(ctx, `DELETE FROM domains WHERE id = $1`, domainID) + + destMailboxID := uuid.New() + destAddress := "realuser@" + domainName + _, err = pool.Exec(ctx, `INSERT INTO mailboxes (id, domain_id, local_part, email_address, password_hash) VALUES ($1, $2, 'realuser', $3, '$argon2id$v=19$m=65536,t=3,p=4$c29tZXNhbHQ$RdescudvJCsgt3ub+b+dWRWJTmaaJObG')`, + destMailboxID, domainID, destAddress) + if err != nil { + t.Fatalf("seed destination mailbox: %v", err) + } + + aliasAddress := "postmaster@" + domainName + _, err = pool.Exec(ctx, `INSERT INTO aliases (id, domain_id, source_address, destination_addresses) VALUES ($1, $2, $3, $4)`, + uuid.New(), domainID, aliasAddress, []string{destAddress}) + if err != nil { + t.Fatalf("seed alias: %v", err) + } + + repo := NewMailboxRepository(pool) + targets, err := repo.ResolveDeliveryTargets(ctx, aliasAddress) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(targets) != 1 || targets[0].ID != destMailboxID { + t.Errorf("targets = %+v, want single record with ID %v (alias destination)", targets, destMailboxID) + } +} + +func TestMailboxRepository_ResolveDeliveryTargets_ResolvesCatchAllAlias(t *testing.T) { + pool := testPool(t) + defer pool.Close() + ctx := context.Background() + + domainID := uuid.New() + domainName := "repo-test-catchall-" + domainID.String() + ".invalid" + _, err := pool.Exec(ctx, `INSERT INTO domains (id, name, punycode_name) VALUES ($1, $2, $3)`, + domainID, domainName, domainName) + if err != nil { + t.Fatalf("seed domain: %v", err) + } + defer pool.Exec(ctx, `DELETE FROM domains WHERE id = $1`, domainID) + + destMailboxID := uuid.New() + destAddress := "catchall-target@" + domainName + _, err = pool.Exec(ctx, `INSERT INTO mailboxes (id, domain_id, local_part, email_address, password_hash) VALUES ($1, $2, 'catchall-target', $3, '$argon2id$v=19$m=65536,t=3,p=4$c29tZXNhbHQ$RdescudvJCsgt3ub+b+dWRWJTmaaJObG')`, + destMailboxID, domainID, destAddress) + if err != nil { + t.Fatalf("seed destination mailbox: %v", err) + } + + _, err = pool.Exec(ctx, `INSERT INTO aliases (id, domain_id, source_address, destination_addresses, is_catch_all) VALUES ($1, $2, $3, $4, true)`, + uuid.New(), domainID, "*@"+domainName, []string{destAddress}) + if err != nil { + t.Fatalf("seed catch-all alias: %v", err) + } + + repo := NewMailboxRepository(pool) + targets, err := repo.ResolveDeliveryTargets(ctx, "whatever-unmapped@"+domainName) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(targets) != 1 || targets[0].ID != destMailboxID { + t.Errorf("targets = %+v, want single record with ID %v (catch-all destination)", targets, destMailboxID) + } +} + +func TestMailboxRepository_ResolveDeliveryTargets_ReturnsEmptySliceWhenNothingMatches(t *testing.T) { + pool := testPool(t) + defer pool.Close() + + repo := NewMailboxRepository(pool) + targets, err := repo.ResolveDeliveryTargets(context.Background(), "definitely-not-seeded@example.invalid") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(targets) != 0 { + t.Errorf("targets = %+v, want empty slice for an address with no mailbox and no alias", targets) + } +} + +func TestMailboxRepository_ResolveDeliveryTargets_SkipsAliasDestinationThatIsNotAnInternalMailbox(t *testing.T) { + pool := testPool(t) + defer pool.Close() + ctx := context.Background() + + domainID := uuid.New() + domainName := "repo-test-external-" + domainID.String() + ".invalid" + _, err := pool.Exec(ctx, `INSERT INTO domains (id, name, punycode_name) VALUES ($1, $2, $3)`, + domainID, domainName, domainName) + if err != nil { + t.Fatalf("seed domain: %v", err) + } + defer pool.Exec(ctx, `DELETE FROM domains WHERE id = $1`, domainID) + + aliasAddress := "forward@" + domainName + _, err = pool.Exec(ctx, `INSERT INTO aliases (id, domain_id, source_address, destination_addresses) VALUES ($1, $2, $3, $4)`, + uuid.New(), domainID, aliasAddress, []string{"someone@external-provider.invalid"}) + if err != nil { + t.Fatalf("seed alias: %v", err) + } + + repo := NewMailboxRepository(pool) + targets, err := repo.ResolveDeliveryTargets(ctx, aliasAddress) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(targets) != 0 { + t.Errorf("targets = %+v, want empty slice - external forwarding destinations are not internal mailboxes and outbound delivery does not exist yet", targets) + } +} diff --git a/services/protocols/internal/presentation/smtp/backend.go b/services/protocols/internal/presentation/smtp/backend.go index ac0e48a..c7e2712 100644 --- a/services/protocols/internal/presentation/smtp/backend.go +++ b/services/protocols/internal/presentation/smtp/backend.go @@ -61,23 +61,35 @@ func (s *session) Rcpt(to string, _ *gosmtp.RcptOptions) error { } } - rec, err := s.useCase.ResolveRecipient(context.Background(), to) + targets, err := s.useCase.ResolveRecipient(context.Background(), to) if err != nil { - if errors.Is(err, usecase.ErrRecipientNotFound) { + switch { + case errors.Is(err, usecase.ErrRecipientNotFound): return &gosmtp.SMTPError{ Code: 550, EnhancedCode: gosmtp.EnhancedCode{5, 1, 1}, Message: "User unknown", } - } - return &gosmtp.SMTPError{ - Code: 451, - EnhancedCode: gosmtp.EnhancedCode{4, 3, 0}, - Message: "Temporary error resolving recipient", + case errors.Is(err, usecase.ErrMailboxQuotaExceeded): + return &gosmtp.SMTPError{ + Code: 452, + EnhancedCode: gosmtp.EnhancedCode{4, 2, 2}, + Message: "Mailbox full", + } + default: + return &gosmtp.SMTPError{ + Code: 451, + EnhancedCode: gosmtp.EnhancedCode{4, 3, 0}, + Message: "Temporary error resolving recipient", + } } } - s.recipients = append(s.recipients, *rec) - s.recipientAddresses = append(s.recipientAddresses, to) + // One RCPT TO can resolve to multiple mailboxes (alias fan-out) - each + // gets its own delivery, all recorded under the same envelope address. + for _, rec := range targets { + s.recipients = append(s.recipients, rec) + s.recipientAddresses = append(s.recipientAddresses, to) + } return nil } diff --git a/services/protocols/internal/presentation/smtp/backend_test.go b/services/protocols/internal/presentation/smtp/backend_test.go index 748df87..b5ef21f 100644 --- a/services/protocols/internal/presentation/smtp/backend_test.go +++ b/services/protocols/internal/presentation/smtp/backend_test.go @@ -26,6 +26,14 @@ func (f *fakeMailboxRepository) FindActiveByAddress(_ context.Context, address s return &rec, nil } +func (f *fakeMailboxRepository) ResolveDeliveryTargets(_ context.Context, address string) ([]port.MailboxRecord, error) { + rec, ok := f.byAddress[address] + if !ok { + return []port.MailboxRecord{}, nil + } + return []port.MailboxRecord{rec}, nil +} + type recordingBlobStorage struct { stored []byte ref port.BlobRef @@ -61,7 +69,7 @@ func newTestSession(mailboxes *fakeMailboxRepository, blobs *recordingBlobStorag func TestSession_Rcpt_AcceptsKnownActiveRecipient(t *testing.T) { mailboxes := &fakeMailboxRepository{byAddress: map[string]port.MailboxRecord{ - "user@example.test": {ID: uuid.New(), MaxMessageBytes: 1000}, + "user@example.test": {ID: uuid.New(), MaxMessageBytes: 1000, QuotaBytes: 1000}, }} session := newTestSession(mailboxes, &recordingBlobStorage{}, &recordingMessageRepository{}) @@ -94,7 +102,7 @@ func TestSession_Rcpt_Rejects550ForUnknownRecipient(t *testing.T) { func TestSession_Data_PersistsOnceForSingleRecipient(t *testing.T) { mailboxID := uuid.New() mailboxes := &fakeMailboxRepository{byAddress: map[string]port.MailboxRecord{ - "user@example.test": {ID: mailboxID, MaxMessageBytes: 1000}, + "user@example.test": {ID: mailboxID, MaxMessageBytes: 1000, QuotaBytes: 1000}, }} blobs := &recordingBlobStorage{ref: port.BlobRef{ID: uuid.New(), SHA256: "abc", SizeBytes: 5}} messages := &recordingMessageRepository{} @@ -119,10 +127,30 @@ func TestSession_Data_PersistsOnceForSingleRecipient(t *testing.T) { } } +func TestSession_Rcpt_Rejects452ForFullMailbox(t *testing.T) { + mailboxes := &fakeMailboxRepository{byAddress: map[string]port.MailboxRecord{ + "full@example.test": {ID: uuid.New(), QuotaBytes: 1000, UsedBytes: 1000}, + }} + session := newTestSession(mailboxes, &recordingBlobStorage{}, &recordingMessageRepository{}) + + _ = session.Mail("sender@example.test", &gosmtp.MailOptions{}) + err := session.Rcpt("full@example.test", &gosmtp.RcptOptions{}) + if err == nil { + t.Fatal("expected an error for a full mailbox, got nil") + } + var smtpErr *gosmtp.SMTPError + if !errors.As(err, &smtpErr) { + t.Fatalf("error is not a *gosmtp.SMTPError: %v", err) + } + if smtpErr.Code != 452 { + t.Errorf("Code = %d, want 452", smtpErr.Code) + } +} + func TestSession_Rcpt_DeduplicatesRepeatedAddress(t *testing.T) { mailboxID := uuid.New() mailboxes := &fakeMailboxRepository{byAddress: map[string]port.MailboxRecord{ - "user@example.test": {ID: mailboxID, MaxMessageBytes: 1000}, + "user@example.test": {ID: mailboxID, MaxMessageBytes: 1000, QuotaBytes: 1000}, }} blobs := &recordingBlobStorage{ref: port.BlobRef{ID: uuid.New(), SHA256: "abc", SizeBytes: 5}} messages := &recordingMessageRepository{} @@ -147,7 +175,7 @@ func TestSession_Rcpt_DeduplicatesRepeatedAddress(t *testing.T) { func TestSession_Reset_ClearsAccumulatedRecipients(t *testing.T) { mailboxes := &fakeMailboxRepository{byAddress: map[string]port.MailboxRecord{ - "user@example.test": {ID: uuid.New(), MaxMessageBytes: 1000}, + "user@example.test": {ID: uuid.New(), MaxMessageBytes: 1000, QuotaBytes: 1000}, }} messages := &recordingMessageRepository{} session := newTestSession(mailboxes, &recordingBlobStorage{}, messages)