#2: implement F1 sub-project 1 - inbound SMTP listener with EML storage
Adds a real :25 SMTP listener to the Go protocols service, giving F1's "Nucleo SMTP" milestone its first working sub-project: application, infrastructure, and presentation layers on top of F0's domain-only Clean Architecture. - ProcessInboundEmailUseCase resolves recipients against Postgres and persists accepted mail (blob dedup by content hash, transactional UID allocation + outbox event, never sending 250 OK before durable commit per PLAN.md section 6.1). - Postgres repositories via pgx/v5 (replacing lib/pq entirely). - Local-disk blob storage with content-hash dedup, rename-before-commit ordering so a DB row can never reference a file that isn't actually on disk. - A local-file TLS cert provider for STARTTLS, with an ephemeral self-signed fallback so a missing/misconfigured cert degrades gracefully instead of crash-looping the service (PLAN.md section 8.4). - A github.com/emersion/go-smtp Backend/Session adapting real SMTP traffic to the use case: unknown recipients get 550 5.1.1 before DATA, duplicate RCPT TO for the same address is deduplicated, TLS is hardened to the project's required PFS-only cipher/curve policy. - A dev-seed script and Makefile target for local testing. Explicitly out of scope (separate follow-up sub-projects): alias resolution, SPF/DKIM/DMARC/ARC verification, Rspamd/ClamAV, Submission ports with SASL auth, DKIM signing, outbound delivery, and the real Traefik acme.json cert watcher. Verified live end-to-end via WSL Docker: the full compose stack (db, redis, migrate, auth, storage, protocols, web) reaches healthy, and a real swaks SMTP session is accepted and durably persisted with correct sequential UIDs, while an unknown recipient is correctly rejected with 550 5.1.1 and zero rows persisted. Built via 10 sequential tasks under subagent-driven development, each independently reviewed, plus a final whole-branch review that caught and fixed a TLS-crash bug, a blob commit-ordering bug, missing TLS hardening, and missing CI integration coverage before merge. Known follow-up, not blocking: email_messages' partition only covers July 2026 and needs a partition-management job before 2026-08-01.
This commit is contained in:
Lucas Rayan committed
1 parent
7c91182111
commit
e501cdc69e
27 files changed
+1468
-17
No files matched your search
@@ -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
|
||||
}
|
||||
+82
@@ -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)
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user