#2: F1 sub-project 1 - inbound SMTP listener with EML storage

Merges the first F1 sub-project: a real :25 SMTP listener for the Go
protocols service with STARTTLS, Postgres-backed recipient resolution,
content-hash-deduplicated blob storage, and transactional persistence.
See commit e501cdc for full details.

.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

.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

.gitignore

@@ -17,6 +17,7 @@ build/
 .env
 .env.local
 acme.json
+local/
 
 # Go
 *.test

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

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:

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;

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

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")
+	ctx := context.Background()
-	db, err := sql.Open("postgres", dbURL)
+
+	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
+}

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
 )

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/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
-github.com/lib/pq v1.10.9/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o=
+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=

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)
+}

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)
+}

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)
+}

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)
+}

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
+}

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")
+	}
+}

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
+}

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)
+	}
+}

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
+}

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")
+	}
+}

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
+}

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)
+	}
+}

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)
+}

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
+}

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")
+	}
+}

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
+}

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))
+	}
+}