diff --git a/.env.example b/.env.example index 9727b26..a75a6e3 100644 --- a/.env.example +++ b/.env.example @@ -21,6 +21,9 @@ PUBLIC_IPV6= # --- TLS (F0 has no real certs — protocols stub reports degraded, never crashes) --- TLS_MODE=traefik COOLIFY_PROXY_DIR=./local/acme +PROTOCOLS_CERT_PATH=/etc/lambdamail/certs/cert.pem +PROTOCOLS_KEY_PATH=/etc/lambdamail/certs/key.pem +PROTOCOLS_SPOOL_DIR=/var/lambdamail/spool # --- Webmail --- WEBMAIL_HOST=localhost:3000 diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 5204f26..5a8a6b5 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -14,15 +14,39 @@ jobs: go-protocols: runs-on: ubuntu-latest + services: + postgres: + image: postgres:16-alpine + env: + POSTGRES_DB: lambdamail_test + POSTGRES_USER: lambdamail_test + POSTGRES_PASSWORD: lambdamail_test + ports: + - 5432:5432 + options: >- + --health-cmd pg_isready + --health-interval 5s + --health-timeout 5s + --health-retries 10 defaults: run: working-directory: services/protocols steps: - uses: actions/checkout@v4 - uses: actions/setup-go@v5 - with: { go-version: "1.22" } + with: { go-version-file: services/protocols/go.mod } + - name: Run migrations + run: | + docker run --rm --network host \ + -v "$(pwd)/../../migrations:/migrations" \ + migrate/migrate:v4.17.1 \ + -path=/migrations \ + -database "postgres://lambdamail_test:lambdamail_test@localhost:5432/lambdamail_test?sslmode=disable" \ + up - run: go vet ./... - run: go test ./... -v + env: + TEST_DATABASE_URL: postgres://lambdamail_test:lambdamail_test@localhost:5432/lambdamail_test?sslmode=disable ts-auth: runs-on: ubuntu-latest diff --git a/.gitignore b/.gitignore index 8646c6c..a0d4f1b 100644 --- a/.gitignore +++ b/.gitignore @@ -17,6 +17,7 @@ build/ .env .env.local acme.json +local/ # Go *.test diff --git a/Makefile b/Makefile index 9bcaf49..9751d69 100644 --- a/Makefile +++ b/Makefile @@ -1,4 +1,4 @@ -.PHONY: up down logs test lint migrate-up migrate-down +.PHONY: up down logs test lint migrate-up migrate-down gen-dev-cert seed up: docker compose --env-file .env up --build @@ -25,3 +25,12 @@ migrate-up: migrate-down: docker run --rm -v $(PWD)/migrations:/migrations --network host \ migrate/migrate -path=/migrations -database "$$DATABASE_URL" down 1 + +gen-dev-cert: + mkdir -p local/certs + openssl req -x509 -newkey ec -pkeyopt ec_paramgen_curve:prime256v1 \ + -keyout local/certs/key.pem -out local/certs/cert.pem \ + -days 365 -nodes -subj "/CN=mail.localhost" + +seed: + docker compose exec -T db psql -U $${POSTGRES_USER:-lambdamail} -d $${POSTGRES_DB:-lambdamail} < scripts/dev-seed.sql diff --git a/docker-compose.yaml b/docker-compose.yaml index 5675d39..06417e1 100644 --- a/docker-compose.yaml +++ b/docker-compose.yaml @@ -92,9 +92,16 @@ services: build: { context: ./services/protocols } restart: unless-stopped logging: *default-logging - ports: ["8080:8080"] + ports: ["8080:8080", "25:25"] environment: DATABASE_URL: postgres://${POSTGRES_USER}:${POSTGRES_PASSWORD}@db:5432/${POSTGRES_DB}?sslmode=disable + PRIMARY_MAIL_HOST: ${PRIMARY_MAIL_HOST:-mail.localhost} + PROTOCOLS_CERT_PATH: /etc/lambdamail/certs/cert.pem + PROTOCOLS_KEY_PATH: /etc/lambdamail/certs/key.pem + PROTOCOLS_SPOOL_DIR: /var/lambdamail/spool + volumes: + - ./local/certs:/etc/lambdamail/certs:ro + - protocols_spool:/var/lambdamail/spool depends_on: db: { condition: service_healthy } migrate: { condition: service_completed_successfully } @@ -127,3 +134,4 @@ services: volumes: postgres_data: redis_data: + protocols_spool: diff --git a/scripts/dev-seed.sql b/scripts/dev-seed.sql new file mode 100644 index 0000000..504743a --- /dev/null +++ b/scripts/dev-seed.sql @@ -0,0 +1,27 @@ +-- Dev-only fixture data for local SMTP testing. Not a schema migration - +-- never applied in production, and safe to re-run (idempotent upserts). +INSERT INTO domains (id, name, punycode_name, is_active) +VALUES ('00000000-0000-0000-0000-000000000001', 'example.test', 'example.test', true) +ON CONFLICT (name) DO NOTHING; + +-- Password: "dev-password-only" hashed with Argon2id - never use in production. +INSERT INTO mailboxes (id, domain_id, local_part, email_address, password_hash, is_active) +VALUES ( + '00000000-0000-0000-0000-000000000002', + '00000000-0000-0000-0000-000000000001', + 'postmaster', + 'postmaster@example.test', + '$argon2id$v=19$m=65536,t=3,p=4$c29tZXNhbHQ$RdescudvJCsgt3ub+b+dWRWJTmaaJObG', + true +) +ON CONFLICT (domain_id, local_part) DO NOTHING; + +INSERT INTO folders (id, mailbox_id, name, special_use, parent_id) +VALUES ( + '00000000-0000-0000-0000-000000000003', + '00000000-0000-0000-0000-000000000002', + 'INBOX', + 'inbox', + NULL +) +ON CONFLICT (mailbox_id, name) WHERE (parent_id IS NULL) DO NOTHING; diff --git a/services/protocols/Dockerfile b/services/protocols/Dockerfile index 4ba0e64..c4ba5d9 100644 --- a/services/protocols/Dockerfile +++ b/services/protocols/Dockerfile @@ -1,4 +1,4 @@ -FROM golang:1.22-alpine AS build +FROM golang:1.25-alpine AS build WORKDIR /src COPY go.mod go.sum ./ RUN go mod download diff --git a/services/protocols/cmd/lambdamail-protocols/main.go b/services/protocols/cmd/lambdamail-protocols/main.go index 6567563..d55c5cb 100644 --- a/services/protocols/cmd/lambdamail-protocols/main.go +++ b/services/protocols/cmd/lambdamail-protocols/main.go @@ -1,30 +1,92 @@ // Command lambdamail-protocols is the entry point for the protocols service. -// F0 boots only the health endpoint; SMTP/IMAP/POP3/ManageSieve listeners land in F1/F2. package main import ( - "database/sql" + "context" + "crypto/tls" "log" "net/http" "os" - _ "github.com/lib/pq" + gosmtp "github.com/emersion/go-smtp" + + "lambdamail/protocols/internal/application/usecase" "lambdamail/protocols/internal/health" + "lambdamail/protocols/internal/infrastructure/diskstorage" + "lambdamail/protocols/internal/infrastructure/postgres" + tlsprovider "lambdamail/protocols/internal/infrastructure/tls" + smtppresentation "lambdamail/protocols/internal/presentation/smtp" ) func main() { - dbURL := os.Getenv("DATABASE_URL") - db, err := sql.Open("postgres", dbURL) + ctx := context.Background() + + pool, err := postgres.NewPool(ctx, os.Getenv("DATABASE_URL")) if err != nil { - log.Fatalf("open database: %v", err) + log.Fatalf("open database pool: %v", err) } - defer db.Close() + defer pool.Close() - handler := health.Handler(func() error { return db.Ping() }) + spoolDir := envOrDefault("PROTOCOLS_SPOOL_DIR", "/var/lambdamail/spool") + if err := os.MkdirAll(spoolDir, 0o755); err != nil { + log.Fatalf("create spool dir %s: %v", spoolDir, err) + } + mailboxes := postgres.NewMailboxRepository(pool) + blobs := diskstorage.NewLocalDiskBlobStorage(pool, spoolDir) + messages := postgres.NewInboundMessageRepository(pool) + useCase := usecase.NewProcessInboundEmailUseCase(mailboxes, blobs, messages) + + certProvider, err := tlsprovider.NewLocalFileCertProvider( + os.Getenv("PROTOCOLS_CERT_PATH"), + os.Getenv("PROTOCOLS_KEY_PATH"), + ) + if err != nil { + log.Printf("WARNING: could not load configured TLS certificate (%v) - falling back to an ephemeral self-signed certificate; this is degraded mode, not suitable for real deployments", err) + certProvider, err = tlsprovider.NewEphemeralSelfSignedCertProvider() + if err != nil { + log.Fatalf("generate fallback self-signed certificate: %v", err) + } + } + + backend := smtppresentation.NewBackend(useCase) + smtpServer := gosmtp.NewServer(backend) + smtpServer.Addr = ":25" + smtpServer.Domain = envOrDefault("PRIMARY_MAIL_HOST", "mail.localhost") + smtpServer.MaxMessageBytes = 52428800 // 50 MB default; per-domain limit is a future iteration + smtpServer.MaxRecipients = 100 + smtpServer.TLSConfig = &tls.Config{ + GetCertificate: certProvider.GetCertificate, + MinVersion: tls.VersionTLS12, + CipherSuites: []uint16{ + tls.TLS_ECDHE_ECDSA_WITH_AES_256_GCM_SHA384, + tls.TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384, + tls.TLS_ECDHE_ECDSA_WITH_CHACHA20_POLY1305, + tls.TLS_ECDHE_RSA_WITH_CHACHA20_POLY1305, + tls.TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256, + tls.TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256, + }, + CurvePreferences: []tls.CurveID{tls.X25519, tls.CurveP256, tls.CurveP384}, + } + + go func() { + log.Printf("lambdamail-protocols SMTP listening on %s", smtpServer.Addr) + if err := smtpServer.ListenAndServe(); err != nil { + log.Fatalf("smtp serve: %v", err) + } + }() + + handler := health.Handler(func() error { return pool.Ping(ctx) }) addr := ":8080" - log.Printf("lambdamail-protocols listening on %s (health only - F0 scaffold)", addr) + log.Printf("lambdamail-protocols health listening on %s", addr) if err := http.ListenAndServe(addr, handler); err != nil { log.Fatalf("serve: %v", err) } } + +func envOrDefault(key, fallback string) string { + if v := os.Getenv(key); v != "" { + return v + } + return fallback +} diff --git a/services/protocols/go.mod b/services/protocols/go.mod index 0e22467..48536cd 100644 --- a/services/protocols/go.mod +++ b/services/protocols/go.mod @@ -1,8 +1,18 @@ module lambdamail/protocols -go 1.22 +go 1.25.0 require ( + github.com/emersion/go-smtp v0.24.0 github.com/google/uuid v1.6.0 - github.com/lib/pq v1.10.9 + github.com/jackc/pgx/v5 v5.10.0 +) + +require ( + github.com/emersion/go-sasl v0.0.0-20241020182733-b788ff22d5a6 // 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 + 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 ae20c4c..a1f0a43 100644 --- a/services/protocols/go.sum +++ b/services/protocols/go.sum @@ -1,4 +1,32 @@ +github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/emersion/go-sasl v0.0.0-20241020182733-b788ff22d5a6 h1:oP4q0fw+fOSWn3DfFi4EXdT+B+gTtzx8GC9xsc26Znk= +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/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/lib/pq v1.10.9 h1:YXG7RB+JIjhP29X+OtkiDnYaXQwpS4JEWq7dtCCRUEw= -github.com/lib/pq v1.10.9/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o= +github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM= +github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg= +github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo= +github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM= +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/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= +github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= +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= +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= +golang.org/x/text v0.29.0/go.mod h1:7MhJOA9CD2qZyOKYazxdYMF85OwPdEr9jTtBpO7ydH4= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/services/protocols/internal/application/port/blob_storage.go b/services/protocols/internal/application/port/blob_storage.go new file mode 100644 index 0000000..98389fa --- /dev/null +++ b/services/protocols/internal/application/port/blob_storage.go @@ -0,0 +1,22 @@ +package port + +import ( + "context" + "io" + + "github.com/google/uuid" +) + +// BlobRef identifies a stored message blob (PLAN.md section 9, message_blobs). +type BlobRef struct { + ID uuid.UUID + SHA256 string + SizeBytes int64 +} + +// BlobStorage persists raw message bytes, deduplicated by content hash. +type BlobStorage interface { + // Store writes r to durable storage and returns its BlobRef. Calling + // Store twice with identical content returns the same BlobRef.ID. + Store(ctx context.Context, r io.Reader) (BlobRef, error) +} diff --git a/services/protocols/internal/application/port/cert_provider.go b/services/protocols/internal/application/port/cert_provider.go new file mode 100644 index 0000000..1ca0b78 --- /dev/null +++ b/services/protocols/internal/application/port/cert_provider.go @@ -0,0 +1,10 @@ +package port + +import "crypto/tls" + +// CertProvider supplies the TLS certificate for STARTTLS. Its signature +// matches tls.Config.GetCertificate exactly so any implementation can be +// wired in directly: tls.Config{GetCertificate: certProvider.GetCertificate}. +type CertProvider interface { + GetCertificate(hello *tls.ClientHelloInfo) (*tls.Certificate, error) +} diff --git a/services/protocols/internal/application/port/inbound_message_repository.go b/services/protocols/internal/application/port/inbound_message_repository.go new file mode 100644 index 0000000..2dbf721 --- /dev/null +++ b/services/protocols/internal/application/port/inbound_message_repository.go @@ -0,0 +1,29 @@ +package port + +import ( + "context" + + "github.com/google/uuid" +) + +// PersistInboundMessageInput is everything needed to durably record one +// recipient's copy of an accepted inbound message. +type PersistInboundMessageInput struct { + MailboxID uuid.UUID + Blob BlobRef + SenderAddress string + RecipientAddress string + SPFResult string + DKIMResult string + DMARCResult string +} + +// InboundMessageRepository durably records an accepted inbound message for +// one recipient: allocates the next folder UID, inserts the email_messages +// row, increments the blob's ref_count, and writes the transactional outbox +// event - all in a single transaction (PLAN.md section 6.1, section 9.4). +type InboundMessageRepository interface { + // Persist returns the allocated IMAP UID on success. The caller may only + // report success to the SMTP client after this returns without error. + Persist(ctx context.Context, input PersistInboundMessageInput) (uid int64, err error) +} diff --git a/services/protocols/internal/application/port/mailbox_repository.go b/services/protocols/internal/application/port/mailbox_repository.go new file mode 100644 index 0000000..4b402fa --- /dev/null +++ b/services/protocols/internal/application/port/mailbox_repository.go @@ -0,0 +1,22 @@ +// Package port declares the interfaces the application layer depends on. +// Zero infrastructure imports: implementations live in internal/infrastructure. +package port + +import ( + "context" + + "github.com/google/uuid" +) + +// MailboxRecord is the subset of mailbox + domain data the inbound flow needs. +type MailboxRecord struct { + ID uuid.UUID + MaxMessageBytes int64 +} + +// MailboxRepository resolves a recipient address to an active mailbox. +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) +} diff --git a/services/protocols/internal/application/usecase/process_inbound_email.go b/services/protocols/internal/application/usecase/process_inbound_email.go new file mode 100644 index 0000000..a367398 --- /dev/null +++ b/services/protocols/internal/application/usecase/process_inbound_email.go @@ -0,0 +1,79 @@ +// Package usecase contains the application's use cases (Clean Architecture layer 2). +// Depends only on internal/domain and internal/application/port - no infrastructure. +package usecase + +import ( + "context" + "errors" + "io" + + "lambdamail/protocols/internal/application/port" +) + +// ErrRecipientNotFound is returned by ResolveRecipient when no active +// 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") + +// 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. +type ProcessInboundEmailInput struct { + Sender string + Recipients []port.MailboxRecord + RecipientAddresses []string + Body io.Reader +} + +// ProcessInboundEmailUseCase implements the inbound half of PLAN.md section +// 3's ProcessInboundEmailUseCase (this sub-project's scope: recipient +// resolution + durable persistence; SPF/DKIM/DMARC/spam/virus checks land in +// later sub-projects). +type ProcessInboundEmailUseCase struct { + mailboxes port.MailboxRepository + blobs port.BlobStorage + messages port.InboundMessageRepository +} + +func NewProcessInboundEmailUseCase(mailboxes port.MailboxRepository, blobs port.BlobStorage, messages port.InboundMessageRepository) *ProcessInboundEmailUseCase { + 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) + if err != nil { + return nil, err + } + if rec == nil { + return nil, ErrRecipientNotFound + } + return rec, nil +} + +// Handle is called once from the SMTP session's DATA handler, after one or +// more successful RCPT TO calls. It stores the message body exactly once +// and persists exactly one email_messages row per recipient, referencing +// the same blob (PLAN.md section 9's dedup design: 1 email to 50 +// recipients = 1 file). +func (uc *ProcessInboundEmailUseCase) Handle(ctx context.Context, input ProcessInboundEmailInput) error { + blob, err := uc.blobs.Store(ctx, input.Body) + if err != nil { + return err + } + + for i, recipient := range input.Recipients { + _, err := uc.messages.Persist(ctx, port.PersistInboundMessageInput{ + MailboxID: recipient.ID, + Blob: blob, + SenderAddress: input.Sender, + RecipientAddress: input.RecipientAddresses[i], + SPFResult: "none", + DKIMResult: "none", + DMARCResult: "none", + }) + if err != nil { + return err + } + } + return nil +} diff --git a/services/protocols/internal/application/usecase/process_inbound_email_test.go b/services/protocols/internal/application/usecase/process_inbound_email_test.go new file mode 100644 index 0000000..2d4f6d8 --- /dev/null +++ b/services/protocols/internal/application/usecase/process_inbound_email_test.go @@ -0,0 +1,127 @@ +package usecase + +import ( + "bytes" + "context" + "errors" + "io" + "testing" + + "github.com/google/uuid" + "lambdamail/protocols/internal/application/port" +) + +type fakeMailboxRepository struct { + byAddress map[string]port.MailboxRecord +} + +func (f *fakeMailboxRepository) FindActiveByAddress(_ context.Context, address string) (*port.MailboxRecord, error) { + rec, ok := f.byAddress[address] + if !ok { + return nil, nil + } + return &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}, + }} + uc := NewProcessInboundEmailUseCase(mailboxes, nil, nil) + + rec, 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) + } +} + +func TestUseCase_ResolveRecipient_ReturnsErrRecipientNotFoundForUnknownAddress(t *testing.T) { + mailboxes := &fakeMailboxRepository{byAddress: map[string]port.MailboxRecord{}} + uc := NewProcessInboundEmailUseCase(mailboxes, nil, nil) + + _, err := uc.ResolveRecipient(context.Background(), "nobody@example.test") + if !errors.Is(err, ErrRecipientNotFound) { + t.Fatalf("error = %v, want ErrRecipientNotFound", err) + } +} + +type stubBlobStorage struct { + ref port.BlobRef + err error +} + +func (s *stubBlobStorage) Store(_ context.Context, r io.Reader) (port.BlobRef, error) { + _, _ = io.Copy(io.Discard, r) + return s.ref, s.err +} + +type recordingMessageRepository struct { + persisted []port.PersistInboundMessageInput + nextUID int64 +} + +func (r *recordingMessageRepository) Persist(_ context.Context, input port.PersistInboundMessageInput) (int64, error) { + r.persisted = append(r.persisted, input) + r.nextUID++ + return r.nextUID, nil +} + +func TestUseCase_Handle_StoresBlobOnceAndPersistsOncePerRecipient(t *testing.T) { + blobRef := port.BlobRef{ID: uuid.New(), SHA256: "abc123", SizeBytes: 42} + blobs := &stubBlobStorage{ref: blobRef} + messages := &recordingMessageRepository{} + uc := NewProcessInboundEmailUseCase(nil, blobs, messages) + + recipients := []port.MailboxRecord{ + {ID: uuid.New(), MaxMessageBytes: 1000}, + {ID: uuid.New(), MaxMessageBytes: 1000}, + } + addresses := []string{"a@example.test", "b@example.test"} + + err := uc.Handle(context.Background(), ProcessInboundEmailInput{ + Sender: "sender@example.test", + Recipients: recipients, + RecipientAddresses: addresses, + Body: bytes.NewBufferString("Subject: hi\r\n\r\nbody"), + }) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + if len(messages.persisted) != 2 { + t.Fatalf("persisted %d messages, want 2 (one per recipient, blob stored once)", len(messages.persisted)) + } + for i, p := range messages.persisted { + if p.Blob.ID != blobRef.ID { + t.Errorf("persisted[%d].Blob.ID = %v, want %v (same blob for every recipient)", i, p.Blob.ID, blobRef.ID) + } + if p.MailboxID != recipients[i].ID { + t.Errorf("persisted[%d].MailboxID = %v, want %v", i, p.MailboxID, recipients[i].ID) + } + if p.RecipientAddress != addresses[i] { + t.Errorf("persisted[%d].RecipientAddress = %q, want %q", i, p.RecipientAddress, addresses[i]) + } + if p.SPFResult != "none" || p.DKIMResult != "none" || p.DMARCResult != "none" { + t.Errorf("persisted[%d] auth results = %q/%q/%q, want \"none\"/\"none\"/\"none\" (verification lands in a later sub-project)", i, p.SPFResult, p.DKIMResult, p.DMARCResult) + } + } +} + +func TestUseCase_Handle_PropagatesBlobStorageError(t *testing.T) { + blobs := &stubBlobStorage{err: errors.New("disk full")} + uc := NewProcessInboundEmailUseCase(nil, blobs, &recordingMessageRepository{}) + + err := uc.Handle(context.Background(), ProcessInboundEmailInput{ + Sender: "sender@example.test", + Recipients: []port.MailboxRecord{{ID: uuid.New()}}, + RecipientAddresses: []string{"a@example.test"}, + Body: bytes.NewBufferString("x"), + }) + if err == nil { + t.Fatal("expected error when blob storage fails, got nil") + } +} diff --git a/services/protocols/internal/infrastructure/diskstorage/local_disk_blob_storage.go b/services/protocols/internal/infrastructure/diskstorage/local_disk_blob_storage.go new file mode 100644 index 0000000..3377ac0 --- /dev/null +++ b/services/protocols/internal/infrastructure/diskstorage/local_disk_blob_storage.go @@ -0,0 +1,101 @@ +// Package diskstorage implements port.BlobStorage on the local filesystem. +package diskstorage + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "fmt" + "io" + "os" + "path/filepath" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" + + "lambdamail/protocols/internal/application/port" +) + +// LocalDiskBlobStorage writes message bodies under baseDir, sharded into +// subdirectories by the first two hex chars of the content hash (avoids +// unbounded single-directory fanout). +type LocalDiskBlobStorage struct { + pool *pgxpool.Pool + baseDir string +} + +func NewLocalDiskBlobStorage(pool *pgxpool.Pool, baseDir string) *LocalDiskBlobStorage { + return &LocalDiskBlobStorage{pool: pool, baseDir: baseDir} +} + +func (s *LocalDiskBlobStorage) Store(ctx context.Context, r io.Reader) (port.BlobRef, error) { + tmp, err := os.CreateTemp(s.baseDir, "incoming-*.tmp") + if err != nil { + return port.BlobRef{}, fmt.Errorf("create temp file: %w", err) + } + tmpPath := tmp.Name() + defer os.Remove(tmpPath) // no-op once renamed to its final path + + hasher := sha256.New() + size, err := io.Copy(io.MultiWriter(tmp, hasher), r) + if err != nil { + tmp.Close() + return port.BlobRef{}, fmt.Errorf("write blob: %w", err) + } + if err := tmp.Sync(); err != nil { + tmp.Close() + return port.BlobRef{}, fmt.Errorf("fsync blob: %w", err) + } + if err := tmp.Close(); err != nil { + return port.BlobRef{}, fmt.Errorf("close blob: %w", err) + } + + sum := hex.EncodeToString(hasher.Sum(nil)) + shardDir := filepath.Join(s.baseDir, sum[:2]) + if err := os.MkdirAll(shardDir, 0o755); err != nil { + return port.BlobRef{}, fmt.Errorf("create shard dir: %w", err) + } + finalPath := filepath.Join(shardDir, sum+".eml") + + // Rename into place BEFORE inserting the DB row, so a committed row + // never references a file that isn't actually there yet (a rename + // failure after an insert would otherwise leave a durable row pointing + // at nothing, and any concurrent Store with identical content could + // dedup onto that row and report success for content that was never + // safely stored). + if _, statErr := os.Stat(finalPath); statErr == nil { + // Content-identical file already stored (same hash => same path, + // and file content is byte-identical); no need to overwrite it. + os.Remove(tmpPath) + } else if !os.IsNotExist(statErr) { + return port.BlobRef{}, fmt.Errorf("stat final blob path: %w", statErr) + } else { + if err := os.Rename(tmpPath, finalPath); err != nil { + return port.BlobRef{}, fmt.Errorf("rename blob into place: %w", err) + } + } + + // Insert-or-fetch: identical content dedups to the existing row. + var id uuid.UUID + err = s.pool.QueryRow(ctx, ` + INSERT INTO message_blobs (content_sha256, storage_driver, storage_path, size_bytes) + VALUES ($1, 'local', $2, $3) + ON CONFLICT (content_sha256) DO NOTHING + RETURNING id + `, sum, finalPath, size).Scan(&id) + if err != nil { + if err == pgx.ErrNoRows { + // Conflict: a row already exists for this hash. + err = s.pool.QueryRow(ctx, `SELECT id FROM message_blobs WHERE content_sha256 = $1`, sum).Scan(&id) + if err != nil { + return port.BlobRef{}, fmt.Errorf("fetch existing blob row: %w", err) + } + // Content already stored on disk from a prior call; discard this duplicate write. + return port.BlobRef{ID: id, SHA256: sum, SizeBytes: size}, nil + } + return port.BlobRef{}, fmt.Errorf("insert blob row: %w", err) + } + + return port.BlobRef{ID: id, SHA256: sum, SizeBytes: size}, nil +} diff --git a/services/protocols/internal/infrastructure/diskstorage/local_disk_blob_storage_test.go b/services/protocols/internal/infrastructure/diskstorage/local_disk_blob_storage_test.go new file mode 100644 index 0000000..aa57a3f --- /dev/null +++ b/services/protocols/internal/infrastructure/diskstorage/local_disk_blob_storage_test.go @@ -0,0 +1,82 @@ +package diskstorage + +import ( + "bytes" + "context" + "os" + "path/filepath" + "testing" + + "github.com/jackc/pgx/v5/pgxpool" +) + +func testPoolForBlobStorage(t *testing.T) *pgxpool.Pool { + t.Helper() + dbURL := os.Getenv("TEST_DATABASE_URL") + if dbURL == "" { + t.Skip("TEST_DATABASE_URL not set - skipping blob storage integration test") + } + pool, err := pgxpool.New(context.Background(), dbURL) + if err != nil { + t.Skipf("cannot connect to TEST_DATABASE_URL: %v", err) + } + if err := pool.Ping(context.Background()); err != nil { + t.Skipf("cannot ping TEST_DATABASE_URL: %v", err) + } + return pool +} + +func TestLocalDiskBlobStorage_Store_WritesFileAndInsertsBlobRow(t *testing.T) { + pool := testPoolForBlobStorage(t) + defer pool.Close() + dir := t.TempDir() + storage := NewLocalDiskBlobStorage(pool, dir) + + content := []byte("Subject: test\r\n\r\nhello world") + ref, err := storage.Store(context.Background(), bytes.NewReader(content)) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + defer pool.Exec(context.Background(), `DELETE FROM message_blobs WHERE id = $1`, ref.ID) + + if ref.SizeBytes != int64(len(content)) { + t.Errorf("SizeBytes = %d, want %d", ref.SizeBytes, len(content)) + } + + entries, err := filepath.Glob(filepath.Join(dir, "*", "*")) + if err != nil { + t.Fatalf("glob: %v", err) + } + if len(entries) != 1 { + t.Fatalf("found %d files under %s, want 1", len(entries), dir) + } + written, err := os.ReadFile(entries[0]) + if err != nil { + t.Fatalf("read written file: %v", err) + } + if !bytes.Equal(written, content) { + t.Error("written file content does not match input") + } +} + +func TestLocalDiskBlobStorage_Store_DedupsIdenticalContent(t *testing.T) { + pool := testPoolForBlobStorage(t) + defer pool.Close() + dir := t.TempDir() + storage := NewLocalDiskBlobStorage(pool, dir) + + content := []byte("identical content for dedup test") + first, err := storage.Store(context.Background(), bytes.NewReader(content)) + if err != nil { + t.Fatalf("first Store: %v", err) + } + defer pool.Exec(context.Background(), `DELETE FROM message_blobs WHERE id = $1`, first.ID) + + second, err := storage.Store(context.Background(), bytes.NewReader(content)) + if err != nil { + t.Fatalf("second Store: %v", err) + } + if second.ID != first.ID { + t.Errorf("second.ID = %v, want %v - identical content must dedup to the same blob (PLAN.md section 9)", second.ID, first.ID) + } +} diff --git a/services/protocols/internal/infrastructure/postgres/inbound_message_repository.go b/services/protocols/internal/infrastructure/postgres/inbound_message_repository.go new file mode 100644 index 0000000..8c3ab90 --- /dev/null +++ b/services/protocols/internal/infrastructure/postgres/inbound_message_repository.go @@ -0,0 +1,75 @@ +package postgres + +import ( + "context" + "fmt" + + "github.com/jackc/pgx/v5/pgxpool" + + "lambdamail/protocols/internal/application/port" +) + +// InboundMessageRepository implements port.InboundMessageRepository against +// Postgres: one transaction per Persist call, matching PLAN.md section 6.1's +// "250 OK only after fsync + COMMIT" rule (the fsync happens in BlobStorage; +// this repository's COMMIT is the second half of that durability guarantee). +type InboundMessageRepository struct { + pool *pgxpool.Pool +} + +func NewInboundMessageRepository(pool *pgxpool.Pool) *InboundMessageRepository { + return &InboundMessageRepository{pool: pool} +} + +func (r *InboundMessageRepository) Persist(ctx context.Context, input port.PersistInboundMessageInput) (int64, error) { + tx, err := r.pool.Begin(ctx) + if err != nil { + return 0, err + } + defer tx.Rollback(ctx) // no-op if Commit succeeds + + var folderID string + var uid int64 + err = tx.QueryRow(ctx, ` + SELECT id, uid_next FROM folders + WHERE mailbox_id = $1 AND special_use = 'inbox' + FOR UPDATE + `, input.MailboxID).Scan(&folderID, &uid) + if err != nil { + 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) + } + + // email_messages.id defaults to gen_random_uuid() per the migration - no explicit value needed. + _, err = tx.Exec(ctx, ` + INSERT INTO email_messages ( + mailbox_id, folder_id, uid, blob_id, + sender_address, recipient_addresses, size_bytes, + spf_result, dkim_result, dmarc_result + ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10) + `, input.MailboxID, folderID, uid, input.Blob.ID, + input.SenderAddress, []string{input.RecipientAddress}, input.Blob.SizeBytes, + input.SPFResult, input.DKIMResult, input.DMARCResult) + if err != nil { + return 0, fmt.Errorf("insert email_messages: %w", err) + } + + if _, err := tx.Exec(ctx, `UPDATE message_blobs SET ref_count = ref_count + 1 WHERE id = $1`, input.Blob.ID); err != nil { + return 0, fmt.Errorf("increment blob ref_count: %w", err) + } + + if _, err := tx.Exec(ctx, ` + INSERT INTO domain_events_outbox (event_type, aggregate_id, payload) + VALUES ('EmailReceived', $1, $2) + `, input.MailboxID, fmt.Sprintf(`{"recipient":%q,"uid":%d}`, input.RecipientAddress, uid)); err != nil { + return 0, fmt.Errorf("insert outbox event: %w", err) + } + + if err := tx.Commit(ctx); err != nil { + return 0, fmt.Errorf("commit: %w", err) + } + return uid, nil +} diff --git a/services/protocols/internal/infrastructure/postgres/inbound_message_repository_test.go b/services/protocols/internal/infrastructure/postgres/inbound_message_repository_test.go new file mode 100644 index 0000000..b4473c9 --- /dev/null +++ b/services/protocols/internal/infrastructure/postgres/inbound_message_repository_test.go @@ -0,0 +1,116 @@ +package postgres + +import ( + "context" + "testing" + + "github.com/google/uuid" + "lambdamail/protocols/internal/application/port" +) + +func TestInboundMessageRepository_Persist_AllocatesSequentialUIDs(t *testing.T) { + pool := testPool(t) + defer pool.Close() + ctx := context.Background() + + domainID := uuid.New() + domainName := "msgrepo-test-" + domainID.String() + ".invalid" + _, err := pool.Exec(ctx, `INSERT INTO domains (id, name, punycode_name) VALUES ($1, $2, $3)`, + domainID, domainName, domainName) + if err != nil { + t.Fatalf("seed domain: %v", err) + } + defer pool.Exec(ctx, `DELETE FROM domains WHERE id = $1`, domainID) + + mailboxID := uuid.New() + _, err = pool.Exec(ctx, `INSERT INTO mailboxes (id, domain_id, local_part, email_address, password_hash) VALUES ($1, $2, 'user', $3, '$argon2id$v=19$m=65536,t=3,p=4$c29tZXNhbHQ$RdescudvJCsgt3ub+b+dWRWJTmaaJObG')`, + mailboxID, domainID, "user@msgrepo-test-"+domainID.String()+".invalid") + if err != nil { + t.Fatalf("seed mailbox: %v", err) + } + + folderID := uuid.New() + _, err = pool.Exec(ctx, `INSERT INTO folders (id, mailbox_id, name, special_use) VALUES ($1, $2, 'INBOX', 'inbox')`, + folderID, mailboxID) + if err != nil { + t.Fatalf("seed folder: %v", err) + } + + blobID := uuid.New() + _, err = pool.Exec(ctx, `INSERT INTO message_blobs (id, content_sha256, storage_driver, storage_path, size_bytes) VALUES ($1, $2, 'local', '/tmp/x', 10)`, + blobID, "deadbeef00000000000000000000000000000000000000000000000000000000"[:64]) + if err != nil { + t.Fatalf("seed blob: %v", err) + } + + repo := NewInboundMessageRepository(pool) + blob := port.BlobRef{ID: blobID, SHA256: "deadbeef", SizeBytes: 10} + + firstUID, err := repo.Persist(ctx, port.PersistInboundMessageInput{ + MailboxID: mailboxID, Blob: blob, + SenderAddress: "sender@example.test", RecipientAddress: "user@msgrepo-test-" + domainID.String() + ".invalid", + SPFResult: "none", DKIMResult: "none", DMARCResult: "none", + }) + if err != nil { + t.Fatalf("first Persist: %v", err) + } + secondUID, err := repo.Persist(ctx, port.PersistInboundMessageInput{ + MailboxID: mailboxID, Blob: blob, + SenderAddress: "sender2@example.test", RecipientAddress: "user@msgrepo-test-" + domainID.String() + ".invalid", + SPFResult: "none", DKIMResult: "none", DMARCResult: "none", + }) + if err != nil { + t.Fatalf("second Persist: %v", err) + } + if secondUID <= firstUID { + t.Errorf("secondUID (%d) must be strictly greater than firstUID (%d) - RFC 3501 section 2.3.1.1", secondUID, firstUID) + } + + var refCount int + if err := pool.QueryRow(ctx, `SELECT ref_count FROM message_blobs WHERE id = $1`, blobID).Scan(&refCount); err != nil { + t.Fatalf("query ref_count: %v", err) + } + if refCount != 2 { + t.Errorf("ref_count = %d, want 2 after two Persist calls referencing the same blob", refCount) + } +} + +func TestInboundMessageRepository_Persist_ReturnsErrorWhenNoInboxFolderExists(t *testing.T) { + pool := testPool(t) + defer pool.Close() + ctx := context.Background() + + domainID := uuid.New() + domainName := "msgrepo-nofolder-" + domainID.String() + ".invalid" + _, err := pool.Exec(ctx, `INSERT INTO domains (id, name, punycode_name) VALUES ($1, $2, $3)`, + domainID, domainName, domainName) + if err != nil { + t.Fatalf("seed domain: %v", err) + } + defer pool.Exec(ctx, `DELETE FROM domains WHERE id = $1`, domainID) + + mailboxID := uuid.New() + _, err = pool.Exec(ctx, `INSERT INTO mailboxes (id, domain_id, local_part, email_address, password_hash) VALUES ($1, $2, 'user', $3, '$argon2id$v=19$m=65536,t=3,p=4$c29tZXNhbHQ$RdescudvJCsgt3ub+b+dWRWJTmaaJObG')`, + mailboxID, domainID, "nofolder@msgrepo-nofolder-"+domainID.String()+".invalid") + if err != nil { + t.Fatalf("seed mailbox: %v", err) + } + // Deliberately no folder row inserted. + + blobID := uuid.New() + _, err = pool.Exec(ctx, `INSERT INTO message_blobs (id, content_sha256, storage_driver, storage_path, size_bytes) VALUES ($1, $2, 'local', '/tmp/x', 10)`, + blobID, "cafebabe00000000000000000000000000000000000000000000000000000000"[:64]) + if err != nil { + t.Fatalf("seed blob: %v", err) + } + + repo := NewInboundMessageRepository(pool) + _, err = repo.Persist(ctx, port.PersistInboundMessageInput{ + MailboxID: mailboxID, Blob: port.BlobRef{ID: blobID, SizeBytes: 10}, + SenderAddress: "sender@example.test", RecipientAddress: "nofolder@example.test", + SPFResult: "none", DKIMResult: "none", DMARCResult: "none", + }) + if err == nil { + t.Fatal("expected an error when the mailbox has no INBOX folder, got nil") + } +} diff --git a/services/protocols/internal/infrastructure/postgres/mailbox_repository.go b/services/protocols/internal/infrastructure/postgres/mailbox_repository.go new file mode 100644 index 0000000..991907b --- /dev/null +++ b/services/protocols/internal/infrastructure/postgres/mailbox_repository.go @@ -0,0 +1,42 @@ +package postgres + +import ( + "context" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" + + "lambdamail/protocols/internal/application/port" +) + +// MailboxRepository implements port.MailboxRepository against Postgres. +type MailboxRepository struct { + pool *pgxpool.Pool +} + +func NewMailboxRepository(pool *pgxpool.Pool) *MailboxRepository { + return &MailboxRepository{pool: pool} +} + +func (r *MailboxRepository) FindActiveByAddress(ctx context.Context, address string) (*port.MailboxRecord, error) { + row := r.pool.QueryRow(ctx, ` + SELECT m.id, d.max_message_bytes + FROM mailboxes m + JOIN domains d ON d.id = m.domain_id + WHERE m.email_address = $1 + AND m.is_active = true + AND d.is_active = true + `, address) + + var rec port.MailboxRecord + var id uuid.UUID + if err := row.Scan(&id, &rec.MaxMessageBytes); err != nil { + if err == pgx.ErrNoRows { + return nil, nil + } + return nil, err + } + rec.ID = id + return &rec, nil +} diff --git a/services/protocols/internal/infrastructure/postgres/mailbox_repository_test.go b/services/protocols/internal/infrastructure/postgres/mailbox_repository_test.go new file mode 100644 index 0000000..cd9d7f8 --- /dev/null +++ b/services/protocols/internal/infrastructure/postgres/mailbox_repository_test.go @@ -0,0 +1,120 @@ +package postgres + +import ( + "context" + "os" + "testing" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgxpool" +) + +// testPool returns a pool for TEST_DATABASE_URL, or skips the test if that +// env var is unset or the database is unreachable - these are integration +// tests against a real Postgres, not unit tests. +func testPool(t *testing.T) *pgxpool.Pool { + t.Helper() + dbURL := os.Getenv("TEST_DATABASE_URL") + if dbURL == "" { + t.Skip("TEST_DATABASE_URL not set - skipping Postgres integration test") + } + pool, err := NewPool(context.Background(), dbURL) + if err != nil { + t.Skipf("cannot connect to TEST_DATABASE_URL: %v", err) + } + if err := pool.Ping(context.Background()); err != nil { + t.Skipf("cannot ping TEST_DATABASE_URL: %v", err) + } + return pool +} + +func TestMailboxRepository_FindActiveByAddress_ReturnsNilForUnknownAddress(t *testing.T) { + pool := testPool(t) + defer pool.Close() + repo := NewMailboxRepository(pool) + + rec, err := repo.FindActiveByAddress(context.Background(), "definitely-not-seeded@example.invalid") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if rec != nil { + t.Errorf("rec = %+v, want nil for an address with no matching mailbox", rec) + } +} + +func TestMailboxRepository_FindActiveByAddress_ReturnsRecordForSeededMailbox(t *testing.T) { + pool := testPool(t) + defer pool.Close() + ctx := context.Background() + + domainID := uuid.New() + domainName := "repo-test-" + domainID.String() + ".invalid" + // NOTE: reusing a single placeholder ($2) for both `name` (CITEXT) and + // `punycode_name` (VARCHAR) makes Postgres's extended-protocol parameter + // type inference fail with "inconsistent types deduced for parameter $2" + // (SQLSTATE 42P08), confirmed via isolated repro against this same DB - + // see task-3-report.md. Using distinct placeholders for the same value + // avoids that false-negative without changing what's inserted. + _, err := pool.Exec(ctx, `INSERT INTO domains (id, name, punycode_name, max_message_bytes) VALUES ($1, $2, $3, 12345)`, + 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@repo-test-" + domainID.String() + ".invalid" + _, 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) + 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 active mailbox") + } + if rec.ID != mailboxID { + t.Errorf("ID = %v, want %v", rec.ID, mailboxID) + } + if rec.MaxMessageBytes != 12345 { + t.Errorf("MaxMessageBytes = %d, want 12345 (from the domain row)", rec.MaxMessageBytes) + } +} + +func TestMailboxRepository_FindActiveByAddress_ExcludesInactiveMailbox(t *testing.T) { + pool := testPool(t) + defer pool.Close() + ctx := context.Background() + + domainID := uuid.New() + domainName := "repo-test-inactive-" + domainID.String() + ".invalid" + // See NOTE above TestMailboxRepository_FindActiveByAddress_ReturnsRecordForSeededMailbox + // re: distinct placeholders for `name`/`punycode_name`. + _, 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@repo-test-inactive-" + domainID.String() + ".invalid" + _, err = pool.Exec(ctx, `INSERT INTO mailboxes (id, domain_id, local_part, email_address, password_hash, is_active) VALUES ($1, $2, 'user', $3, '$argon2id$v=19$m=65536,t=3,p=4$c29tZXNhbHQ$RdescudvJCsgt3ub+b+dWRWJTmaaJObG', false)`, + 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.Errorf("rec = %+v, want nil - mailbox is_active=false must be excluded", rec) + } +} diff --git a/services/protocols/internal/infrastructure/postgres/pool.go b/services/protocols/internal/infrastructure/postgres/pool.go new file mode 100644 index 0000000..887d995 --- /dev/null +++ b/services/protocols/internal/infrastructure/postgres/pool.go @@ -0,0 +1,14 @@ +// Package postgres contains pgx-based implementations of the application's +// repository ports (PLAN.md section 3's ADR: pgx for Go repositories). +package postgres + +import ( + "context" + + "github.com/jackc/pgx/v5/pgxpool" +) + +// NewPool builds a connection pool for databaseURL (a postgres:// URL). +func NewPool(ctx context.Context, databaseURL string) (*pgxpool.Pool, error) { + return pgxpool.New(ctx, databaseURL) +} diff --git a/services/protocols/internal/infrastructure/tls/local_file_cert_provider.go b/services/protocols/internal/infrastructure/tls/local_file_cert_provider.go new file mode 100644 index 0000000..472f24a --- /dev/null +++ b/services/protocols/internal/infrastructure/tls/local_file_cert_provider.go @@ -0,0 +1,62 @@ +// Package tlsprovider implements port.CertProvider by loading a static +// cert/key pair from disk - suitable for local development. The Traefik +// acme.json directory watcher (PLAN.md section 8) is a separate, later +// adapter behind the same port.CertProvider interface. +package tlsprovider + +import ( + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/tls" + "crypto/x509" + "crypto/x509/pkix" + "fmt" + "math/big" + "time" +) + +// LocalFileCertProvider serves one certificate loaded once at construction. +type LocalFileCertProvider struct { + cert tls.Certificate +} + +func NewLocalFileCertProvider(certPath, keyPath string) (*LocalFileCertProvider, error) { + cert, err := tls.LoadX509KeyPair(certPath, keyPath) + if err != nil { + return nil, fmt.Errorf("load cert/key pair (cert=%s, key=%s): %w", certPath, keyPath, err) + } + return &LocalFileCertProvider{cert: cert}, nil +} + +func (p *LocalFileCertProvider) GetCertificate(_ *tls.ClientHelloInfo) (*tls.Certificate, error) { + return &p.cert, nil +} + +// NewEphemeralSelfSignedCertProvider generates a throwaway self-signed +// certificate in memory. Used as a degraded-mode fallback when no real +// certificate is configured or loadable (PLAN.md section 8.4: never crash +// on a missing cert - serve degraded instead). +func NewEphemeralSelfSignedCertProvider() (*LocalFileCertProvider, error) { + priv, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + if err != nil { + return nil, fmt.Errorf("generate ephemeral key: %w", err) + } + template := x509.Certificate{ + SerialNumber: big.NewInt(1), + Subject: pkix.Name{CommonName: "ephemeral-self-signed.invalid"}, + NotBefore: time.Now(), + NotAfter: time.Now().Add(24 * time.Hour), + KeyUsage: x509.KeyUsageDigitalSignature, + ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth}, + } + der, err := x509.CreateCertificate(rand.Reader, &template, &template, &priv.PublicKey, priv) + if err != nil { + return nil, fmt.Errorf("create ephemeral certificate: %w", err) + } + cert := tls.Certificate{ + Certificate: [][]byte{der}, + PrivateKey: priv, + } + return &LocalFileCertProvider{cert: cert}, nil +} diff --git a/services/protocols/internal/infrastructure/tls/local_file_cert_provider_test.go b/services/protocols/internal/infrastructure/tls/local_file_cert_provider_test.go new file mode 100644 index 0000000..277a1d8 --- /dev/null +++ b/services/protocols/internal/infrastructure/tls/local_file_cert_provider_test.go @@ -0,0 +1,104 @@ +package tlsprovider + +import ( + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/tls" + "crypto/x509" + "crypto/x509/pkix" + "encoding/pem" + "math/big" + "os" + "path/filepath" + "testing" + "time" +) + +// generateSelfSignedPEMForTest returns a minimal self-signed cert/key pair +// generated with Go's crypto/x509, so this test has no external dependency +// on openssl being installed. +func generateSelfSignedPEMForTest(t *testing.T) (certPEM, keyPEM []byte) { + t.Helper() + priv, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + if err != nil { + t.Fatalf("generate key: %v", err) + } + template := x509.Certificate{ + SerialNumber: big.NewInt(1), + Subject: pkix.Name{CommonName: "test.local"}, + NotBefore: time.Now(), + NotAfter: time.Now().Add(time.Hour), + KeyUsage: x509.KeyUsageDigitalSignature, + ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth}, + } + der, err := x509.CreateCertificate(rand.Reader, &template, &template, &priv.PublicKey, priv) + if err != nil { + t.Fatalf("create certificate: %v", err) + } + certPEM = pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der}) + keyDER, err := x509.MarshalECPrivateKey(priv) + if err != nil { + t.Fatalf("marshal key: %v", err) + } + keyPEM = pem.EncodeToMemory(&pem.Block{Type: "EC PRIVATE KEY", Bytes: keyDER}) + return certPEM, keyPEM +} + +// writeTestCertAndKey writes a self-signed cert/key pair to disk for testing +func writeTestCertAndKey(t *testing.T, dir string) (certPath, keyPath string) { + t.Helper() + certPEM, keyPEM := generateSelfSignedPEMForTest(t) + certPath = filepath.Join(dir, "cert.pem") + keyPath = filepath.Join(dir, "key.pem") + if err := os.WriteFile(certPath, certPEM, 0o600); err != nil { + t.Fatalf("write cert: %v", err) + } + if err := os.WriteFile(keyPath, keyPEM, 0o600); err != nil { + t.Fatalf("write key: %v", err) + } + return certPath, keyPath +} + +func TestNewLocalFileCertProvider_LoadsValidCertAndKey(t *testing.T) { + dir := t.TempDir() + certPath, keyPath := writeTestCertAndKey(t, dir) + + provider, err := NewLocalFileCertProvider(certPath, keyPath) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + cert, err := provider.GetCertificate(&tls.ClientHelloInfo{}) + if err != nil { + t.Fatalf("GetCertificate error: %v", err) + } + if cert == nil { + t.Fatal("GetCertificate returned nil cert") + } +} + +func TestNewLocalFileCertProvider_FailsFastOnMissingFiles(t *testing.T) { + _, err := NewLocalFileCertProvider("/nonexistent/cert.pem", "/nonexistent/key.pem") + if err == nil { + t.Fatal("expected error for nonexistent cert/key paths, got nil") + } +} + +func TestNewEphemeralSelfSignedCertProvider_GeneratesUsableCert(t *testing.T) { + provider, err := NewEphemeralSelfSignedCertProvider() + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + cert, err := provider.GetCertificate(&tls.ClientHelloInfo{}) + if err != nil { + t.Fatalf("GetCertificate error: %v", err) + } + if cert == nil { + t.Fatal("GetCertificate returned nil cert") + } + if len(cert.Certificate) == 0 { + t.Fatal("expected at least one certificate DER blob") + } +} diff --git a/services/protocols/internal/presentation/smtp/backend.go b/services/protocols/internal/presentation/smtp/backend.go new file mode 100644 index 0000000..ac0e48a --- /dev/null +++ b/services/protocols/internal/presentation/smtp/backend.go @@ -0,0 +1,106 @@ +// Package smtppresentation adapts github.com/emersion/go-smtp's Backend/ +// Session interfaces to ProcessInboundEmailUseCase (Clean Architecture +// layer 4 - PLAN.md section 3). +package smtppresentation + +import ( + "context" + "errors" + "io" + + gosmtp "github.com/emersion/go-smtp" + + "lambdamail/protocols/internal/application/port" + "lambdamail/protocols/internal/application/usecase" +) + +// Backend is the go-smtp entry point: one Session per connection. +type Backend struct { + useCase *usecase.ProcessInboundEmailUseCase +} + +func NewBackend(uc *usecase.ProcessInboundEmailUseCase) *Backend { + return &Backend{useCase: uc} +} + +func (b *Backend) NewSession(_ *gosmtp.Conn) (gosmtp.Session, error) { + return &session{useCase: b.useCase}, nil +} + +// session holds one SMTP transaction's accumulated state (RFC 5321 section +// 4.1.1: MAIL, then one or more RCPT, then DATA, then reset). +type session struct { + useCase *usecase.ProcessInboundEmailUseCase + from string + recipients []port.MailboxRecord + recipientAddresses []string +} + +func (s *session) Reset() { + s.from = "" + s.recipients = nil + s.recipientAddresses = nil +} + +func (s *session) Logout() error { + return nil +} + +func (s *session) Mail(from string, _ *gosmtp.MailOptions) error { + s.from = from + return nil +} + +func (s *session) Rcpt(to string, _ *gosmtp.RcptOptions) error { + for _, existing := range s.recipientAddresses { + if existing == to { + // Duplicate RCPT TO for an address already accepted in this + // transaction (some MTAs retry): accept silently without + // persisting a second delivery. + return nil + } + } + + rec, err := s.useCase.ResolveRecipient(context.Background(), to) + if err != nil { + if 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", + } + } + s.recipients = append(s.recipients, *rec) + s.recipientAddresses = append(s.recipientAddresses, to) + return nil +} + +func (s *session) Data(r io.Reader) error { + if len(s.recipients) == 0 { + return &gosmtp.SMTPError{ + Code: 503, + EnhancedCode: gosmtp.EnhancedCode{5, 5, 1}, + Message: "RCPT TO required before DATA", + } + } + err := s.useCase.Handle(context.Background(), usecase.ProcessInboundEmailInput{ + Sender: s.from, + Recipients: s.recipients, + RecipientAddresses: s.recipientAddresses, + Body: r, + }) + if err != nil { + return &gosmtp.SMTPError{ + Code: 451, + EnhancedCode: gosmtp.EnhancedCode{4, 3, 0}, + Message: "Temporary error storing message", + } + } + return nil +} diff --git a/services/protocols/internal/presentation/smtp/backend_test.go b/services/protocols/internal/presentation/smtp/backend_test.go new file mode 100644 index 0000000..748df87 --- /dev/null +++ b/services/protocols/internal/presentation/smtp/backend_test.go @@ -0,0 +1,168 @@ +package smtppresentation + +import ( + "bytes" + "context" + "errors" + "io" + "testing" + + gosmtp "github.com/emersion/go-smtp" + "github.com/google/uuid" + + "lambdamail/protocols/internal/application/port" + "lambdamail/protocols/internal/application/usecase" +) + +type fakeMailboxRepository struct { + byAddress map[string]port.MailboxRecord +} + +func (f *fakeMailboxRepository) FindActiveByAddress(_ context.Context, address string) (*port.MailboxRecord, error) { + rec, ok := f.byAddress[address] + if !ok { + return nil, nil + } + return &rec, nil +} + +type recordingBlobStorage struct { + stored []byte + ref port.BlobRef +} + +func (b *recordingBlobStorage) Store(_ context.Context, r io.Reader) (port.BlobRef, error) { + data, err := io.ReadAll(r) + if err != nil { + return port.BlobRef{}, err + } + b.stored = data + return b.ref, nil +} + +type recordingMessageRepository struct { + persisted []port.PersistInboundMessageInput +} + +func (m *recordingMessageRepository) Persist(_ context.Context, input port.PersistInboundMessageInput) (int64, error) { + m.persisted = append(m.persisted, input) + return int64(len(m.persisted)), nil +} + +func newTestSession(mailboxes *fakeMailboxRepository, blobs *recordingBlobStorage, messages *recordingMessageRepository) gosmtp.Session { + uc := usecase.NewProcessInboundEmailUseCase(mailboxes, blobs, messages) + backend := NewBackend(uc) + session, err := backend.NewSession(nil) + if err != nil { + panic(err) + } + return session +} + +func TestSession_Rcpt_AcceptsKnownActiveRecipient(t *testing.T) { + mailboxes := &fakeMailboxRepository{byAddress: map[string]port.MailboxRecord{ + "user@example.test": {ID: uuid.New(), MaxMessageBytes: 1000}, + }} + session := newTestSession(mailboxes, &recordingBlobStorage{}, &recordingMessageRepository{}) + + if err := session.Mail("sender@example.test", &gosmtp.MailOptions{}); err != nil { + t.Fatalf("Mail: %v", err) + } + if err := session.Rcpt("user@example.test", &gosmtp.RcptOptions{}); err != nil { + t.Fatalf("Rcpt: %v", err) + } +} + +func TestSession_Rcpt_Rejects550ForUnknownRecipient(t *testing.T) { + mailboxes := &fakeMailboxRepository{byAddress: map[string]port.MailboxRecord{}} + session := newTestSession(mailboxes, &recordingBlobStorage{}, &recordingMessageRepository{}) + + _ = session.Mail("sender@example.test", &gosmtp.MailOptions{}) + err := session.Rcpt("nobody@example.test", &gosmtp.RcptOptions{}) + if err == nil { + t.Fatal("expected an error for an unknown recipient, got nil") + } + var smtpErr *gosmtp.SMTPError + if !errors.As(err, &smtpErr) { + t.Fatalf("error is not a *gosmtp.SMTPError: %v", err) + } + if smtpErr.Code != 550 { + t.Errorf("Code = %d, want 550", smtpErr.Code) + } +} + +func TestSession_Data_PersistsOnceForSingleRecipient(t *testing.T) { + mailboxID := uuid.New() + mailboxes := &fakeMailboxRepository{byAddress: map[string]port.MailboxRecord{ + "user@example.test": {ID: mailboxID, MaxMessageBytes: 1000}, + }} + blobs := &recordingBlobStorage{ref: port.BlobRef{ID: uuid.New(), SHA256: "abc", SizeBytes: 5}} + messages := &recordingMessageRepository{} + session := newTestSession(mailboxes, blobs, messages) + + _ = session.Mail("sender@example.test", &gosmtp.MailOptions{}) + _ = session.Rcpt("user@example.test", &gosmtp.RcptOptions{}) + + body := "Subject: hi\r\n\r\nhello" + if err := session.Data(bytes.NewBufferString(body)); err != nil { + t.Fatalf("Data: %v", err) + } + + if string(blobs.stored) != body { + t.Errorf("stored blob = %q, want %q", blobs.stored, body) + } + if len(messages.persisted) != 1 { + t.Fatalf("persisted %d messages, want 1", len(messages.persisted)) + } + if messages.persisted[0].MailboxID != mailboxID { + t.Errorf("persisted MailboxID = %v, want %v", messages.persisted[0].MailboxID, mailboxID) + } +} + +func TestSession_Rcpt_DeduplicatesRepeatedAddress(t *testing.T) { + mailboxID := uuid.New() + mailboxes := &fakeMailboxRepository{byAddress: map[string]port.MailboxRecord{ + "user@example.test": {ID: mailboxID, MaxMessageBytes: 1000}, + }} + blobs := &recordingBlobStorage{ref: port.BlobRef{ID: uuid.New(), SHA256: "abc", SizeBytes: 5}} + messages := &recordingMessageRepository{} + session := newTestSession(mailboxes, blobs, messages) + + _ = session.Mail("sender@example.test", &gosmtp.MailOptions{}) + if err := session.Rcpt("user@example.test", &gosmtp.RcptOptions{}); err != nil { + t.Fatalf("first Rcpt: %v", err) + } + if err := session.Rcpt("user@example.test", &gosmtp.RcptOptions{}); err != nil { + t.Fatalf("duplicate Rcpt: %v", err) + } + + if err := session.Data(bytes.NewBufferString("Subject: hi\r\n\r\nhello")); err != nil { + t.Fatalf("Data: %v", err) + } + + if len(messages.persisted) != 1 { + t.Fatalf("persisted %d messages, want 1 (duplicate RCPT should not double-persist)", len(messages.persisted)) + } +} + +func TestSession_Reset_ClearsAccumulatedRecipients(t *testing.T) { + mailboxes := &fakeMailboxRepository{byAddress: map[string]port.MailboxRecord{ + "user@example.test": {ID: uuid.New(), MaxMessageBytes: 1000}, + }} + messages := &recordingMessageRepository{} + session := newTestSession(mailboxes, &recordingBlobStorage{}, messages) + + _ = session.Mail("sender@example.test", &gosmtp.MailOptions{}) + _ = session.Rcpt("user@example.test", &gosmtp.RcptOptions{}) + session.Reset() + + // After Reset, Data with no prior Mail/Rcpt in this transaction must not + // silently persist anything. + err := session.Data(bytes.NewBufferString("Subject: x\r\n\r\nbody")) + if err == nil { + t.Fatal("expected an error calling Data after Reset with no recipients, got nil") + } + if len(messages.persisted) != 0 { + t.Errorf("persisted %d messages after Reset, want 0", len(messages.persisted)) + } +}